From d796422c544d3728b7ee484407f1d8c78d698d51 Mon Sep 17 00:00:00 2001 From: Philipp Hancke Date: Fri, 24 Apr 2026 12:56:49 +0200 Subject: [PATCH 1/3] Implement SPED (STUN Protocol for Embedding DTLS) Part of pion/webrtc#3335 --- agent.go | 214 ++++++++++++++++++++++++++++++++++++++++++++++++++ agent_test.go | 57 ++++++++++++++ selection.go | 56 +++++++++++-- 3 files changed, 321 insertions(+), 6 deletions(-) diff --git a/agent.go b/agent.go index 5350d1d2..dfb85467 100644 --- a/agent.go +++ b/agent.go @@ -8,6 +8,7 @@ package ice import ( "context" "fmt" + "hash/crc32" "math" "net" "net/netip" @@ -38,6 +39,32 @@ type bindingRequest struct { nominationValue *uint32 // Tracks nomination value for renomination requests } +type packetWithCrc struct { + data []byte + crc uint32 +} + +type piggybackingState int + +const ( + PiggybackingStateTentative = iota + PiggybackingStateConfirmed + PiggybackingStatePending + PiggybackingStateComplete + PiggybackingStateOff +) + +// DTLS-in-STUN controller. +type piggybackingController struct { + mu sync.Mutex + state piggybackingState + packets []packetWithCrc + packetsIndex int + acks []uint32 + dtlsCallback func(packet []byte, rAddr net.Addr) + newFlight bool +} + // Agent represents the ICE agent. type Agent struct { loop *taskloop.Loop @@ -178,6 +205,8 @@ type Agent struct { lastRenominationTime time.Time turnClientFactory func(*turn.ClientConfig) (turnClient, error) + + piggyback piggybackingController } // NewAgent creates a new Agent. @@ -221,6 +250,13 @@ func newAgentFromConfig(config *AgentConfig, opts ...AgentOption) (*Agent, error agent.addressRewriteRules = rules } + // Embedding DTLS in STUN. This is off by default and enabled + // by the use of `SetDtlsCallback`. + agent.piggyback.mu.Lock() + agent.piggyback.acks = []uint32{} + agent.piggyback.state = PiggybackingStateOff + agent.piggyback.mu.Unlock() + return newAgentWithConfig(agent, opts...) } @@ -755,6 +791,22 @@ func (a *Agent) updateConnectionState(newState ConnectionState) { a.deleteAllCandidates() } + var packetsToFlush []packetWithCrc + a.piggyback.mu.Lock() + if newState == ConnectionStateConnected && a.piggyback.state == PiggybackingStateOff { + // Piggybacking was discovered as not supported. + // Flush any pending DTLS packets. + packetsToFlush = a.piggyback.packets + a.piggyback.packets = []packetWithCrc{} + } + a.piggyback.mu.Unlock() + + if pair := a.getSelectedPair(); pair != nil && len(packetsToFlush) > 0 { + for _, p := range packetsToFlush { + _, _ = pair.Write(p.data) + } + } + a.log.Infof("Setting new connection state: %s", newState) a.connectionState = newState a.connectionStateNotifier.EnqueueConnectionState(newState) @@ -1582,6 +1634,14 @@ func (a *Agent) sendBindingSuccess(m *stun.Message, local, remote Candidate) { Port: port, }, } + if packet, acks := a.GetPiggybackDataAndAcks(); acks != nil { + if acks != nil { + attributes = append(attributes, DtlsInStunAckAttribute(acks)) + } + if packet != nil { + attributes = append(attributes, DtlsInStunAttribute(packet)) + } + } attributes = append(attributes, stun.NewShortTermIntegrity(a.localPwd), stun.Fingerprint) @@ -1842,6 +1902,152 @@ func (a *Agent) getSelectedPair() *CandidatePair { return nil } +// SetDtlsCallback sets the callback for DTLS packets. Setting this callback +// initializes state of the piggybacking state machine to "tentative", i.e. +// expecting embedded packets. +func (a *Agent) SetDtlsCallback(cb func(packet []byte, rAddr net.Addr)) { + a.piggyback.mu.Lock() + defer a.piggyback.mu.Unlock() + a.piggyback.dtlsCallback = cb + if cb != nil { + a.piggyback.state = PiggybackingStateTentative + } +} + +// Piggyback stores a packet to be picked in a round-robin fashion. +// Returns `true` if packet is to be consumed. +func (a *Agent) Piggyback(packet []byte, end bool) bool { + a.piggyback.mu.Lock() + defer a.piggyback.mu.Unlock() + if a.piggyback.state == PiggybackingStateOff { + return a.connectionState != ConnectionStateConnected + } + + if packet != nil { + // If we receive a packet after the end of a flight we need + // to clear the outgoing list. + if a.piggyback.newFlight { + a.piggyback.packets = []packetWithCrc{} + } + a.piggyback.newFlight = end + crc := crc32.ChecksumIEEE(packet) + a.piggyback.packets = append(a.piggyback.packets, packetWithCrc{packet, crc}) + } else { + a.piggyback.state = PiggybackingStatePending + } + // If we are connected we could send DTLS plain. + return true // a.connectionState == ConnectionStateConnected +} + +// GetPiggybackDataAndAcks returns a packet from the stored list in a round-robin fashion and a list of acks. +func (a *Agent) GetPiggybackDataAndAcks() ([]byte, []uint32) { + a.piggyback.mu.Lock() + defer a.piggyback.mu.Unlock() + + if a.piggyback.state == PiggybackingStateOff || a.piggyback.state == PiggybackingStateComplete { + return nil, nil + } + if len(a.piggyback.packets) == 0 { + return nil, a.piggyback.acks + } + + packet := a.piggyback.packets[a.piggyback.packetsIndex] + a.piggyback.packetsIndex = (a.piggyback.packetsIndex + 1) % len(a.piggyback.packets) + + // Return a copy to prevent external modification of the internal buffer + result := make([]byte, len(packet.data)) + copy(result, packet.data) + + return result, a.piggyback.acks +} + +func (a *Agent) ReportPiggybacking(packet []byte, acks []uint32, rAddr net.Addr) { //nolint:cyclop + a.piggyback.mu.Lock() + + if a.piggyback.state == PiggybackingStateComplete || a.piggyback.state == PiggybackingStateOff { + a.piggyback.mu.Unlock() + + return + } + if packet == nil && acks == nil && a.piggyback.state == PiggybackingStateTentative { + // Any pending packets will be flushed later when the ICE connection gets established. + a.log.Infof("Piggybacking discovered as not supported, falling back to normal state") + a.piggyback.dtlsCallback = nil + a.piggyback.state = PiggybackingStateOff + a.piggyback.mu.Unlock() + + return + } + if packet == nil && acks == nil && a.piggyback.acks != nil { + a.log.Infof("Done with the SPED handshake", a.piggyback.state) + a.piggyback.acks = nil + a.piggyback.state = PiggybackingStateComplete + a.piggyback.mu.Unlock() + + return + } + if a.piggyback.state == PiggybackingStateTentative { + a.piggyback.state = PiggybackingStateConfirmed + } + // Handle incoming acks. + if size := len(acks); size > 0 { + beforeLen := len(a.piggyback.packets) + a.piggyback.packets = slices.DeleteFunc(a.piggyback.packets, func(p packetWithCrc) bool { + // Remove packets that were acknowledged. + return slices.Contains(acks, p.crc) + }) + removed := beforeLen - len(a.piggyback.packets) + + // Adjust the index if it's out of bounds after deletion + if a.piggyback.packetsIndex >= removed { + a.piggyback.packetsIndex -= removed + } else { + a.piggyback.packetsIndex = 0 + } + } + if len(packet) == 0 { + a.piggyback.acks = []uint32{} + } + + var dtlsCallback func(packet []byte, rAddr net.Addr) + // Handle the incoming packet. Calculate and store the crc32 of the packet + // for acks, then notify the DTLS packet. + if a.piggyback.dtlsCallback != nil && len(packet) > 0 { + crc := crc32.ChecksumIEEE(packet) + if !slices.Contains(a.piggyback.acks, crc) { + a.piggyback.acks = append(a.piggyback.acks, crc) + if len(a.piggyback.acks) > 4 { + a.piggyback.acks = a.piggyback.acks[1:] + } + } + dtlsCallback = a.piggyback.dtlsCallback + } + + a.piggyback.mu.Unlock() + + if dtlsCallback != nil { + dtlsCallback(packet, rAddr) + } +} + +func (a *Agent) ReportDtlsPacket(packet []byte) { + a.piggyback.mu.Lock() + + if a.piggyback.state == PiggybackingStateComplete || a.piggyback.state == PiggybackingStateOff { + a.piggyback.mu.Unlock() + + return + } + crc := crc32.ChecksumIEEE(packet) + if !slices.Contains(a.piggyback.acks, crc) { + a.piggyback.acks = append(a.piggyback.acks, crc) + if len(a.piggyback.acks) > 4 { + a.piggyback.acks = a.piggyback.acks[1:] + } + } + a.piggyback.mu.Unlock() +} + func (a *Agent) closeMulticastConn() { if a.mDNSConn != nil { if err := a.mDNSConn.Close(); err != nil { @@ -2053,6 +2259,14 @@ func (a *Agent) sendNominationRequest(pair *CandidatePair, nominationValue uint3 a.log.Tracef("Sending renomination request from %s to %s with nomination value %d", pair.Local, pair.Remote, nominationValue) } + if packet, acks := a.GetPiggybackDataAndAcks(); acks != nil { + if acks != nil { + attributes = append(attributes, DtlsInStunAckAttribute(acks)) + } + if packet != nil { + attributes = append(attributes, DtlsInStunAttribute(packet)) + } + } attributes = append(attributes, stun.NewShortTermIntegrity(a.remotePwd), diff --git a/agent_test.go b/agent_test.go index 18f10b03..0fd21fbc 100644 --- a/agent_test.go +++ b/agent_test.go @@ -3386,3 +3386,60 @@ func TestMDNSLocalAddressFromTCPMux(t *testing.T) { require.Nil(t, mDNSLocalAddressFromTCPMux(&stubTCPMux{}, []NetworkType{NetworkTypeTCP4})) }) } + +func TestSped(t *testing.T) { + defer test.CheckRoutines(t)() + + t.Run("Basic embedding", func(t *testing.T) { + aNotifier, aConnected := onConnected() + aAgent, err := NewAgent(&AgentConfig{ + NetworkTypes: supportedNetworkTypes(), + }) + require.NoError(t, err) + require.NoError(t, aAgent.OnConnectionStateChange(aNotifier)) + + var toA string + fromA := "Hello from A" + aAgent.SetDtlsCallback(func(packet []byte, rAddr net.Addr) { + toA = string(packet) + }) + require.True(t, aAgent.Piggyback([]byte(fromA), true)) + + bNotifier, bConnected := onConnected() + bAgent, err := NewAgent(&AgentConfig{ + NetworkTypes: supportedNetworkTypes(), + }) + require.NoError(t, err) + require.NoError(t, bAgent.OnConnectionStateChange(bNotifier)) + + var toB string + fromB := "Hello from B" + bAgent.SetDtlsCallback(func(packet []byte, rAddr net.Addr) { + toB = string(packet) + }) + require.True(t, bAgent.Piggyback([]byte(fromB), true)) + + gatherAndExchangeCandidates(t, aAgent, bAgent) + go func() { + bUfrag, bPwd, err := bAgent.GetLocalUserCredentials() + require.NoError(t, err) + _, err = aAgent.Accept(context.TODO(), bUfrag, bPwd) + require.NoError(t, err) + }() + + go func() { + aUfrag, aPwd, err := aAgent.GetLocalUserCredentials() + require.NoError(t, err) + _, err = bAgent.Dial(context.TODO(), aUfrag, aPwd) + require.NoError(t, err) + }() + + <-aConnected + <-bConnected + require.NoError(t, aAgent.Close()) + require.NoError(t, bAgent.Close()) + + require.Equal(t, toA, fromB) + require.Equal(t, toB, fromA) + }) +} diff --git a/selection.go b/selection.go index 0ebd8ced..75c89b1c 100644 --- a/selection.go +++ b/selection.go @@ -26,6 +26,14 @@ type controllingSelector struct { log logging.LeveledLogger } +func reportPiggybacking(agent *Agent, message *stun.Message, remote Candidate) { + var dtls DtlsInStunAttribute + _ = dtls.GetFrom(message) + var ack DtlsInStunAckAttribute + _ = ack.GetFrom(message) + agent.ReportPiggybacking(dtls, ack, remote.addr()) +} + func (s *controllingSelector) Start() { s.startTime = time.Now() s.nominatedPair = nil @@ -92,6 +100,14 @@ func (s *controllingSelector) nominatePair(pair *CandidatePair) { AttrControlling(s.agent.tieBreaker), PriorityAttr(pair.Local.Priority()), } + if packet, acks := s.agent.GetPiggybackDataAndAcks(); acks != nil { + if acks != nil { + attributes = append(attributes, DtlsInStunAckAttribute(acks)) + } + if packet != nil { + attributes = append(attributes, DtlsInStunAttribute(packet)) + } + } attributes = append(attributes, stun.NewShortTermIntegrity(s.agent.remotePwd), stun.Fingerprint) @@ -107,6 +123,8 @@ func (s *controllingSelector) nominatePair(pair *CandidatePair) { } func (s *controllingSelector) HandleBindingRequest(message *stun.Message, local, remote Candidate) { //nolint:cyclop + reportPiggybacking(s.agent, message, remote) + s.agent.sendBindingSuccess(message, local, remote) pair := s.agent.findPair(local, remote) @@ -141,10 +159,12 @@ func (s *controllingSelector) HandleBindingRequest(message *stun.Message, local, } } -func (s *controllingSelector) HandleSuccessResponse(m *stun.Message, local, remote Candidate, remoteAddr net.Addr) { - ok, pendingRequest, rtt := s.agent.handleInboundBindingSuccess(m.TransactionID) +func (s *controllingSelector) HandleSuccessResponse(message *stun.Message, local, remote Candidate, + remoteAddr net.Addr, +) { + ok, pendingRequest, rtt := s.agent.handleInboundBindingSuccess(message.TransactionID) if !ok { - s.log.Warnf("Discard success response from (%s), unknown TransactionID 0x%x", remote, m.TransactionID) + s.log.Warnf("Discard success response from (%s), unknown TransactionID 0x%x", remote, message.TransactionID) return } @@ -163,6 +183,8 @@ func (s *controllingSelector) HandleSuccessResponse(m *stun.Message, local, remo return } + reportPiggybacking(s.agent, message, remote) + s.log.Tracef("Inbound STUN (SuccessResponse) from %s to %s", remote, local) pair := s.agent.findPair(local, remote) @@ -202,6 +224,14 @@ func (s *controllingSelector) PingCandidate(local, remote Candidate) { AttrControlling(s.agent.tieBreaker), PriorityAttr(local.Priority()), } + if packet, acks := s.agent.GetPiggybackDataAndAcks(); acks != nil { + if acks != nil { + attributes = append(attributes, DtlsInStunAckAttribute(acks)) + } + if packet != nil { + attributes = append(attributes, DtlsInStunAttribute(packet)) + } + } attributes = append(attributes, stun.NewShortTermIntegrity(s.agent.remotePwd), stun.Fingerprint) @@ -354,6 +384,14 @@ func (s *controlledSelector) PingCandidate(local, remote Candidate) { AttrControlled(s.agent.tieBreaker), PriorityAttr(local.Priority()), } + if packet, acks := s.agent.GetPiggybackDataAndAcks(); acks != nil { + if acks != nil { + attributes = append(attributes, DtlsInStunAckAttribute(acks)) + } + if packet != nil { + attributes = append(attributes, DtlsInStunAttribute(packet)) + } + } attributes = append(attributes, stun.NewShortTermIntegrity(s.agent.remotePwd), stun.Fingerprint) @@ -368,7 +406,9 @@ func (s *controlledSelector) PingCandidate(local, remote Candidate) { s.agent.sendBindingRequest(msg, local, remote) } -func (s *controlledSelector) HandleSuccessResponse(m *stun.Message, local, remote Candidate, remoteAddr net.Addr) { +func (s *controlledSelector) HandleSuccessResponse(message *stun.Message, local, remote Candidate, + remoteAddr net.Addr, +) { //nolint:godox // TODO according to the standard we should specifically answer a failed nomination: // https://tools.ietf.org/html/rfc8445#section-7.3.1.5 @@ -377,9 +417,9 @@ func (s *controlledSelector) HandleSuccessResponse(m *stun.Message, local, remot // request with an appropriate error code response (e.g., 400) // [RFC5389]. - ok, pendingRequest, rtt := s.agent.handleInboundBindingSuccess(m.TransactionID) + ok, pendingRequest, rtt := s.agent.handleInboundBindingSuccess(message.TransactionID) if !ok { - s.log.Warnf("Discard message from (%s), unknown TransactionID 0x%x", remote, m.TransactionID) + s.log.Warnf("Discard message from (%s), unknown TransactionID 0x%x", remote, message.TransactionID) return } @@ -421,9 +461,13 @@ func (s *controlledSelector) HandleSuccessResponse(m *stun.Message, local, remot } pair.UpdateRoundTripTime(rtt) + + reportPiggybacking(s.agent, message, remote) } func (s *controlledSelector) HandleBindingRequest(message *stun.Message, local, remote Candidate) { //nolint:cyclop + reportPiggybacking(s.agent, message, remote) + pair := s.agent.findPair(local, remote) if pair == nil { pair = s.agent.addPair(local, remote) From cff123470069f8b5d0d27d5d128a91e94490d542 Mon Sep 17 00:00:00 2001 From: Philipp Hancke Date: Mon, 4 May 2026 14:54:48 +0200 Subject: [PATCH 2/3] Move some sped/piggyback things to separate files --- agent.go | 173 ------------------------------------------- agent_test.go | 57 -------------- piggyback.go | 185 ++++++++++++++++++++++++++++++++++++++++++++++ piggyback_test.go | 72 ++++++++++++++++++ 4 files changed, 257 insertions(+), 230 deletions(-) create mode 100644 piggyback.go create mode 100644 piggyback_test.go diff --git a/agent.go b/agent.go index dfb85467..9333b49f 100644 --- a/agent.go +++ b/agent.go @@ -8,7 +8,6 @@ package ice import ( "context" "fmt" - "hash/crc32" "math" "net" "net/netip" @@ -39,32 +38,6 @@ type bindingRequest struct { nominationValue *uint32 // Tracks nomination value for renomination requests } -type packetWithCrc struct { - data []byte - crc uint32 -} - -type piggybackingState int - -const ( - PiggybackingStateTentative = iota - PiggybackingStateConfirmed - PiggybackingStatePending - PiggybackingStateComplete - PiggybackingStateOff -) - -// DTLS-in-STUN controller. -type piggybackingController struct { - mu sync.Mutex - state piggybackingState - packets []packetWithCrc - packetsIndex int - acks []uint32 - dtlsCallback func(packet []byte, rAddr net.Addr) - newFlight bool -} - // Agent represents the ICE agent. type Agent struct { loop *taskloop.Loop @@ -1902,152 +1875,6 @@ func (a *Agent) getSelectedPair() *CandidatePair { return nil } -// SetDtlsCallback sets the callback for DTLS packets. Setting this callback -// initializes state of the piggybacking state machine to "tentative", i.e. -// expecting embedded packets. -func (a *Agent) SetDtlsCallback(cb func(packet []byte, rAddr net.Addr)) { - a.piggyback.mu.Lock() - defer a.piggyback.mu.Unlock() - a.piggyback.dtlsCallback = cb - if cb != nil { - a.piggyback.state = PiggybackingStateTentative - } -} - -// Piggyback stores a packet to be picked in a round-robin fashion. -// Returns `true` if packet is to be consumed. -func (a *Agent) Piggyback(packet []byte, end bool) bool { - a.piggyback.mu.Lock() - defer a.piggyback.mu.Unlock() - if a.piggyback.state == PiggybackingStateOff { - return a.connectionState != ConnectionStateConnected - } - - if packet != nil { - // If we receive a packet after the end of a flight we need - // to clear the outgoing list. - if a.piggyback.newFlight { - a.piggyback.packets = []packetWithCrc{} - } - a.piggyback.newFlight = end - crc := crc32.ChecksumIEEE(packet) - a.piggyback.packets = append(a.piggyback.packets, packetWithCrc{packet, crc}) - } else { - a.piggyback.state = PiggybackingStatePending - } - // If we are connected we could send DTLS plain. - return true // a.connectionState == ConnectionStateConnected -} - -// GetPiggybackDataAndAcks returns a packet from the stored list in a round-robin fashion and a list of acks. -func (a *Agent) GetPiggybackDataAndAcks() ([]byte, []uint32) { - a.piggyback.mu.Lock() - defer a.piggyback.mu.Unlock() - - if a.piggyback.state == PiggybackingStateOff || a.piggyback.state == PiggybackingStateComplete { - return nil, nil - } - if len(a.piggyback.packets) == 0 { - return nil, a.piggyback.acks - } - - packet := a.piggyback.packets[a.piggyback.packetsIndex] - a.piggyback.packetsIndex = (a.piggyback.packetsIndex + 1) % len(a.piggyback.packets) - - // Return a copy to prevent external modification of the internal buffer - result := make([]byte, len(packet.data)) - copy(result, packet.data) - - return result, a.piggyback.acks -} - -func (a *Agent) ReportPiggybacking(packet []byte, acks []uint32, rAddr net.Addr) { //nolint:cyclop - a.piggyback.mu.Lock() - - if a.piggyback.state == PiggybackingStateComplete || a.piggyback.state == PiggybackingStateOff { - a.piggyback.mu.Unlock() - - return - } - if packet == nil && acks == nil && a.piggyback.state == PiggybackingStateTentative { - // Any pending packets will be flushed later when the ICE connection gets established. - a.log.Infof("Piggybacking discovered as not supported, falling back to normal state") - a.piggyback.dtlsCallback = nil - a.piggyback.state = PiggybackingStateOff - a.piggyback.mu.Unlock() - - return - } - if packet == nil && acks == nil && a.piggyback.acks != nil { - a.log.Infof("Done with the SPED handshake", a.piggyback.state) - a.piggyback.acks = nil - a.piggyback.state = PiggybackingStateComplete - a.piggyback.mu.Unlock() - - return - } - if a.piggyback.state == PiggybackingStateTentative { - a.piggyback.state = PiggybackingStateConfirmed - } - // Handle incoming acks. - if size := len(acks); size > 0 { - beforeLen := len(a.piggyback.packets) - a.piggyback.packets = slices.DeleteFunc(a.piggyback.packets, func(p packetWithCrc) bool { - // Remove packets that were acknowledged. - return slices.Contains(acks, p.crc) - }) - removed := beforeLen - len(a.piggyback.packets) - - // Adjust the index if it's out of bounds after deletion - if a.piggyback.packetsIndex >= removed { - a.piggyback.packetsIndex -= removed - } else { - a.piggyback.packetsIndex = 0 - } - } - if len(packet) == 0 { - a.piggyback.acks = []uint32{} - } - - var dtlsCallback func(packet []byte, rAddr net.Addr) - // Handle the incoming packet. Calculate and store the crc32 of the packet - // for acks, then notify the DTLS packet. - if a.piggyback.dtlsCallback != nil && len(packet) > 0 { - crc := crc32.ChecksumIEEE(packet) - if !slices.Contains(a.piggyback.acks, crc) { - a.piggyback.acks = append(a.piggyback.acks, crc) - if len(a.piggyback.acks) > 4 { - a.piggyback.acks = a.piggyback.acks[1:] - } - } - dtlsCallback = a.piggyback.dtlsCallback - } - - a.piggyback.mu.Unlock() - - if dtlsCallback != nil { - dtlsCallback(packet, rAddr) - } -} - -func (a *Agent) ReportDtlsPacket(packet []byte) { - a.piggyback.mu.Lock() - - if a.piggyback.state == PiggybackingStateComplete || a.piggyback.state == PiggybackingStateOff { - a.piggyback.mu.Unlock() - - return - } - crc := crc32.ChecksumIEEE(packet) - if !slices.Contains(a.piggyback.acks, crc) { - a.piggyback.acks = append(a.piggyback.acks, crc) - if len(a.piggyback.acks) > 4 { - a.piggyback.acks = a.piggyback.acks[1:] - } - } - a.piggyback.mu.Unlock() -} - func (a *Agent) closeMulticastConn() { if a.mDNSConn != nil { if err := a.mDNSConn.Close(); err != nil { diff --git a/agent_test.go b/agent_test.go index 0fd21fbc..18f10b03 100644 --- a/agent_test.go +++ b/agent_test.go @@ -3386,60 +3386,3 @@ func TestMDNSLocalAddressFromTCPMux(t *testing.T) { require.Nil(t, mDNSLocalAddressFromTCPMux(&stubTCPMux{}, []NetworkType{NetworkTypeTCP4})) }) } - -func TestSped(t *testing.T) { - defer test.CheckRoutines(t)() - - t.Run("Basic embedding", func(t *testing.T) { - aNotifier, aConnected := onConnected() - aAgent, err := NewAgent(&AgentConfig{ - NetworkTypes: supportedNetworkTypes(), - }) - require.NoError(t, err) - require.NoError(t, aAgent.OnConnectionStateChange(aNotifier)) - - var toA string - fromA := "Hello from A" - aAgent.SetDtlsCallback(func(packet []byte, rAddr net.Addr) { - toA = string(packet) - }) - require.True(t, aAgent.Piggyback([]byte(fromA), true)) - - bNotifier, bConnected := onConnected() - bAgent, err := NewAgent(&AgentConfig{ - NetworkTypes: supportedNetworkTypes(), - }) - require.NoError(t, err) - require.NoError(t, bAgent.OnConnectionStateChange(bNotifier)) - - var toB string - fromB := "Hello from B" - bAgent.SetDtlsCallback(func(packet []byte, rAddr net.Addr) { - toB = string(packet) - }) - require.True(t, bAgent.Piggyback([]byte(fromB), true)) - - gatherAndExchangeCandidates(t, aAgent, bAgent) - go func() { - bUfrag, bPwd, err := bAgent.GetLocalUserCredentials() - require.NoError(t, err) - _, err = aAgent.Accept(context.TODO(), bUfrag, bPwd) - require.NoError(t, err) - }() - - go func() { - aUfrag, aPwd, err := aAgent.GetLocalUserCredentials() - require.NoError(t, err) - _, err = bAgent.Dial(context.TODO(), aUfrag, aPwd) - require.NoError(t, err) - }() - - <-aConnected - <-bConnected - require.NoError(t, aAgent.Close()) - require.NoError(t, bAgent.Close()) - - require.Equal(t, toA, fromB) - require.Equal(t, toB, fromA) - }) -} diff --git a/piggyback.go b/piggyback.go new file mode 100644 index 00000000..725af013 --- /dev/null +++ b/piggyback.go @@ -0,0 +1,185 @@ +// SPDX-FileCopyrightText: 2026 The Pion community +// SPDX-License-Identifier: MIT + +// Package ice implements the Interactive Connectivity Establishment (ICE) +// protocol defined in rfc5245. +package ice + +import ( + "hash/crc32" + "net" + "slices" + "sync" +) + +type packetWithCrc struct { + data []byte + crc uint32 +} + +type piggybackingState int + +const ( + PiggybackingStateTentative = iota + PiggybackingStateConfirmed + PiggybackingStatePending + PiggybackingStateComplete + PiggybackingStateOff +) + +// DTLS-in-STUN controller. +type piggybackingController struct { + mu sync.Mutex + state piggybackingState + packets []packetWithCrc + packetsIndex int + acks []uint32 + dtlsCallback func(packet []byte, rAddr net.Addr) + newFlight bool +} + +// SetDtlsCallback sets the callback for DTLS packets. Setting this callback +// initializes state of the piggybacking state machine to "tentative", i.e. +// expecting embedded packets. +func (a *Agent) SetDtlsCallback(cb func(packet []byte, rAddr net.Addr)) { + a.piggyback.mu.Lock() + defer a.piggyback.mu.Unlock() + a.piggyback.dtlsCallback = cb + if cb != nil { + a.piggyback.state = PiggybackingStateTentative + } +} + +// Piggyback stores a packet to be picked in a round-robin fashion. +// Returns `true` if packet is to be consumed. +func (a *Agent) Piggyback(packet []byte, end bool) bool { + a.piggyback.mu.Lock() + defer a.piggyback.mu.Unlock() + if a.piggyback.state == PiggybackingStateOff { + return a.connectionState != ConnectionStateConnected + } + + if packet != nil { + // If we receive a packet after the end of a flight we need + // to clear the outgoing list. + if a.piggyback.newFlight { + a.piggyback.packets = []packetWithCrc{} + } + a.piggyback.newFlight = end + crc := crc32.ChecksumIEEE(packet) + a.piggyback.packets = append(a.piggyback.packets, packetWithCrc{packet, crc}) + } else { + a.piggyback.state = PiggybackingStatePending + } + // If we are connected we could send DTLS plain. + return true // a.connectionState == ConnectionStateConnected +} + +// GetPiggybackDataAndAcks returns a packet from the stored list in a round-robin fashion and a list of acks. +func (a *Agent) GetPiggybackDataAndAcks() ([]byte, []uint32) { + a.piggyback.mu.Lock() + defer a.piggyback.mu.Unlock() + + if a.piggyback.state == PiggybackingStateOff || a.piggyback.state == PiggybackingStateComplete { + return nil, nil + } + if len(a.piggyback.packets) == 0 { + return nil, a.piggyback.acks + } + + packet := a.piggyback.packets[a.piggyback.packetsIndex] + a.piggyback.packetsIndex = (a.piggyback.packetsIndex + 1) % len(a.piggyback.packets) + + // Return a copy to prevent external modification of the internal buffer + result := make([]byte, len(packet.data)) + copy(result, packet.data) + + return result, a.piggyback.acks +} + +func (a *Agent) ReportPiggybacking(packet []byte, acks []uint32, rAddr net.Addr) { //nolint:cyclop + a.piggyback.mu.Lock() + + if a.piggyback.state == PiggybackingStateComplete || a.piggyback.state == PiggybackingStateOff { + a.piggyback.mu.Unlock() + + return + } + if packet == nil && acks == nil && a.piggyback.state == PiggybackingStateTentative { + // Any pending packets will be flushed later when the ICE connection gets established. + a.log.Infof("Piggybacking discovered as not supported, falling back to normal state") + a.piggyback.dtlsCallback = nil + a.piggyback.state = PiggybackingStateOff + a.piggyback.mu.Unlock() + + return + } + if packet == nil && acks == nil && a.piggyback.acks != nil { + a.log.Infof("Done with the SPED handshake", a.piggyback.state) + a.piggyback.acks = nil + a.piggyback.state = PiggybackingStateComplete + a.piggyback.mu.Unlock() + + return + } + if a.piggyback.state == PiggybackingStateTentative { + a.piggyback.state = PiggybackingStateConfirmed + } + // Handle incoming acks. + if size := len(acks); size > 0 { + beforeLen := len(a.piggyback.packets) + a.piggyback.packets = slices.DeleteFunc(a.piggyback.packets, func(p packetWithCrc) bool { + // Remove packets that were acknowledged. + return slices.Contains(acks, p.crc) + }) + removed := beforeLen - len(a.piggyback.packets) + + // Adjust the index if it's out of bounds after deletion + if a.piggyback.packetsIndex >= removed { + a.piggyback.packetsIndex -= removed + } else { + a.piggyback.packetsIndex = 0 + } + } + if len(packet) == 0 { + a.piggyback.acks = []uint32{} + } + + var dtlsCallback func(packet []byte, rAddr net.Addr) + // Handle the incoming packet. Calculate and store the crc32 of the packet + // for acks, then notify the DTLS packet. + if a.piggyback.dtlsCallback != nil && len(packet) > 0 { + crc := crc32.ChecksumIEEE(packet) + if !slices.Contains(a.piggyback.acks, crc) { + a.piggyback.acks = append(a.piggyback.acks, crc) + if len(a.piggyback.acks) > 4 { + a.piggyback.acks = a.piggyback.acks[1:] + } + } + dtlsCallback = a.piggyback.dtlsCallback + } + + a.piggyback.mu.Unlock() + + if dtlsCallback != nil { + dtlsCallback(packet, rAddr) + } +} + +func (a *Agent) ReportDtlsPacket(packet []byte) { + a.piggyback.mu.Lock() + + if a.piggyback.state == PiggybackingStateComplete || a.piggyback.state == PiggybackingStateOff { + a.piggyback.mu.Unlock() + + return + } + crc := crc32.ChecksumIEEE(packet) + if !slices.Contains(a.piggyback.acks, crc) { + a.piggyback.acks = append(a.piggyback.acks, crc) + if len(a.piggyback.acks) > 4 { + a.piggyback.acks = a.piggyback.acks[1:] + } + } + a.piggyback.mu.Unlock() +} diff --git a/piggyback_test.go b/piggyback_test.go new file mode 100644 index 00000000..291ecf7c --- /dev/null +++ b/piggyback_test.go @@ -0,0 +1,72 @@ +// SPDX-FileCopyrightText: 2026 The Pion community +// SPDX-License-Identifier: MIT + +//go:build !js + +package ice + +import ( + "context" + "net" + "testing" + + "github.com/pion/transport/v4/test" + "github.com/stretchr/testify/require" +) + +func TestSped(t *testing.T) { + defer test.CheckRoutines(t)() + + t.Run("Basic embedding", func(t *testing.T) { + aNotifier, aConnected := onConnected() + aAgent, err := NewAgent(&AgentConfig{ + NetworkTypes: supportedNetworkTypes(), + }) + require.NoError(t, err) + require.NoError(t, aAgent.OnConnectionStateChange(aNotifier)) + + var toA string + fromA := "Hello from A" + aAgent.SetDtlsCallback(func(packet []byte, rAddr net.Addr) { + toA = string(packet) + }) + require.True(t, aAgent.Piggyback([]byte(fromA), true)) + + bNotifier, bConnected := onConnected() + bAgent, err := NewAgent(&AgentConfig{ + NetworkTypes: supportedNetworkTypes(), + }) + require.NoError(t, err) + require.NoError(t, bAgent.OnConnectionStateChange(bNotifier)) + + var toB string + fromB := "Hello from B" + bAgent.SetDtlsCallback(func(packet []byte, rAddr net.Addr) { + toB = string(packet) + }) + require.True(t, bAgent.Piggyback([]byte(fromB), true)) + + gatherAndExchangeCandidates(t, aAgent, bAgent) + go func() { + bUfrag, bPwd, err := bAgent.GetLocalUserCredentials() + require.NoError(t, err) + _, err = aAgent.Accept(context.TODO(), bUfrag, bPwd) + require.NoError(t, err) + }() + + go func() { + aUfrag, aPwd, err := aAgent.GetLocalUserCredentials() + require.NoError(t, err) + _, err = bAgent.Dial(context.TODO(), aUfrag, aPwd) + require.NoError(t, err) + }() + + <-aConnected + <-bConnected + require.NoError(t, aAgent.Close()) + require.NoError(t, bAgent.Close()) + + require.Equal(t, toA, fromB) + require.Equal(t, toB, fromA) + }) +} From 95838f765ec70f14f9002652a37bb362cad38448 Mon Sep 17 00:00:00 2001 From: Philipp Hancke Date: Fri, 15 May 2026 19:34:43 +0200 Subject: [PATCH 3/3] Move things around --- agent.go | 77 ++++++++++++++++++++-------------------------------- piggyback.go | 53 +++++++++++++++++++++++++++++++++++- selection.go | 43 +++++------------------------ 3 files changed, 88 insertions(+), 85 deletions(-) diff --git a/agent.go b/agent.go index 9333b49f..e14fa12d 100644 --- a/agent.go +++ b/agent.go @@ -225,10 +225,7 @@ func newAgentFromConfig(config *AgentConfig, opts ...AgentOption) (*Agent, error // Embedding DTLS in STUN. This is off by default and enabled // by the use of `SetDtlsCallback`. - agent.piggyback.mu.Lock() - agent.piggyback.acks = []uint32{} - agent.piggyback.state = PiggybackingStateOff - agent.piggyback.mu.Unlock() + agent.piggyback.init() return newAgentWithConfig(agent, opts...) } @@ -753,37 +750,35 @@ func (a *Agent) connectivityChecks() { //nolint:cyclop } func (a *Agent) updateConnectionState(newState ConnectionState) { - if a.connectionState != newState { - // Connection has gone to failed, release all gathered candidates - if newState == ConnectionStateFailed { - a.removeUfragFromMux() - a.checklist = make([]*CandidatePair, 0) - a.pairsByID = make(map[uint64]*CandidatePair) - a.pendingBindingRequests = make([]bindingRequest, 0) - a.setSelectedPair(nil) - a.deleteAllCandidates() - } - - var packetsToFlush []packetWithCrc - a.piggyback.mu.Lock() - if newState == ConnectionStateConnected && a.piggyback.state == PiggybackingStateOff { - // Piggybacking was discovered as not supported. - // Flush any pending DTLS packets. - packetsToFlush = a.piggyback.packets - a.piggyback.packets = []packetWithCrc{} - } - a.piggyback.mu.Unlock() - - if pair := a.getSelectedPair(); pair != nil && len(packetsToFlush) > 0 { - for _, p := range packetsToFlush { - _, _ = pair.Write(p.data) + if a.connectionState == newState { + return + } + + // Connection has gone to failed, release all gathered candidates + if newState == ConnectionStateFailed { + a.removeUfragFromMux() + a.checklist = make([]*CandidatePair, 0) + a.pairsByID = make(map[uint64]*CandidatePair) + a.pendingBindingRequests = make([]bindingRequest, 0) + a.setSelectedPair(nil) + a.deleteAllCandidates() + } + + if newState == ConnectionStateConnected { + // If piggybacking has been discovered as not supported + // flush any pending DTLS packets. + if packets := a.piggyback.flushOnConnected(); len(packets) > 0 { + if pair := a.getSelectedPair(); pair != nil { + for _, p := range packets { + _, _ = pair.Write(p.data) + } } } - - a.log.Infof("Setting new connection state: %s", newState) - a.connectionState = newState - a.connectionStateNotifier.EnqueueConnectionState(newState) } + + a.log.Infof("Setting new connection state: %s", newState) + a.connectionState = newState + a.connectionStateNotifier.EnqueueConnectionState(newState) } func (a *Agent) setSelectedPair(pair *CandidatePair) { @@ -1607,14 +1602,7 @@ func (a *Agent) sendBindingSuccess(m *stun.Message, local, remote Candidate) { Port: port, }, } - if packet, acks := a.GetPiggybackDataAndAcks(); acks != nil { - if acks != nil { - attributes = append(attributes, DtlsInStunAckAttribute(acks)) - } - if packet != nil { - attributes = append(attributes, DtlsInStunAttribute(packet)) - } - } + attributes = a.appendPiggybackAttributes(attributes) attributes = append(attributes, stun.NewShortTermIntegrity(a.localPwd), stun.Fingerprint) @@ -2086,14 +2074,7 @@ func (a *Agent) sendNominationRequest(pair *CandidatePair, nominationValue uint3 a.log.Tracef("Sending renomination request from %s to %s with nomination value %d", pair.Local, pair.Remote, nominationValue) } - if packet, acks := a.GetPiggybackDataAndAcks(); acks != nil { - if acks != nil { - attributes = append(attributes, DtlsInStunAckAttribute(acks)) - } - if packet != nil { - attributes = append(attributes, DtlsInStunAttribute(packet)) - } - } + attributes = a.appendPiggybackAttributes(attributes) attributes = append(attributes, stun.NewShortTermIntegrity(a.remotePwd), diff --git a/piggyback.go b/piggyback.go index 725af013..4bb349a5 100644 --- a/piggyback.go +++ b/piggyback.go @@ -10,6 +10,8 @@ import ( "net" "slices" "sync" + + "github.com/pion/stun/v3" ) type packetWithCrc struct { @@ -20,7 +22,7 @@ type packetWithCrc struct { type piggybackingState int const ( - PiggybackingStateTentative = iota + PiggybackingStateTentative piggybackingState = iota PiggybackingStateConfirmed PiggybackingStatePending PiggybackingStateComplete @@ -38,6 +40,29 @@ type piggybackingController struct { newFlight bool } +// init sets the controller to its initial off state. SetDtlsCallback flips it +// to tentative when piggybacking is enabled. +func (p *piggybackingController) init() { + p.mu.Lock() + defer p.mu.Unlock() + p.acks = []uint32{} + p.state = PiggybackingStateOff +} + +// flushOnConnected returns any pending packets that need to be sent as plain +// DTLS once the ICE connection is established with piggybacking disabled. +func (p *piggybackingController) flushOnConnected() []packetWithCrc { + p.mu.Lock() + defer p.mu.Unlock() + if p.state != PiggybackingStateOff { + return nil + } + packets := p.packets + p.packets = []packetWithCrc{} + + return packets +} + // SetDtlsCallback sets the callback for DTLS packets. Setting this callback // initializes state of the piggybacking state machine to "tentative", i.e. // expecting embedded packets. @@ -166,6 +191,32 @@ func (a *Agent) ReportPiggybacking(packet []byte, acks []uint32, rAddr net.Addr) } } +// appendPiggybackAttributes appends DTLS-in-STUN and ACK attributes (when +// available) to the given setter slice. It is the single place that knows +// the wire-order of those attributes in outgoing STUN messages. +func (a *Agent) appendPiggybackAttributes(attrs []stun.Setter) []stun.Setter { + packet, acks := a.GetPiggybackDataAndAcks() + if acks == nil { + return attrs + } + attrs = append(attrs, DtlsInStunAckAttribute(acks)) + if packet != nil { + attrs = append(attrs, DtlsInStunAttribute(packet)) + } + + return attrs +} + +// reportPiggybackingFromMessage extracts the DTLS-in-STUN payload and ACK list +// from a STUN message and forwards them to the controller. +func (a *Agent) reportPiggybackingFromMessage(message *stun.Message, remote Candidate) { + var dtls DtlsInStunAttribute + _ = dtls.GetFrom(message) + var ack DtlsInStunAckAttribute + _ = ack.GetFrom(message) + a.ReportPiggybacking(dtls, ack, remote.addr()) +} + func (a *Agent) ReportDtlsPacket(packet []byte) { a.piggyback.mu.Lock() diff --git a/selection.go b/selection.go index 75c89b1c..83624c44 100644 --- a/selection.go +++ b/selection.go @@ -26,14 +26,6 @@ type controllingSelector struct { log logging.LeveledLogger } -func reportPiggybacking(agent *Agent, message *stun.Message, remote Candidate) { - var dtls DtlsInStunAttribute - _ = dtls.GetFrom(message) - var ack DtlsInStunAckAttribute - _ = ack.GetFrom(message) - agent.ReportPiggybacking(dtls, ack, remote.addr()) -} - func (s *controllingSelector) Start() { s.startTime = time.Now() s.nominatedPair = nil @@ -100,14 +92,7 @@ func (s *controllingSelector) nominatePair(pair *CandidatePair) { AttrControlling(s.agent.tieBreaker), PriorityAttr(pair.Local.Priority()), } - if packet, acks := s.agent.GetPiggybackDataAndAcks(); acks != nil { - if acks != nil { - attributes = append(attributes, DtlsInStunAckAttribute(acks)) - } - if packet != nil { - attributes = append(attributes, DtlsInStunAttribute(packet)) - } - } + attributes = s.agent.appendPiggybackAttributes(attributes) attributes = append(attributes, stun.NewShortTermIntegrity(s.agent.remotePwd), stun.Fingerprint) @@ -123,7 +108,7 @@ func (s *controllingSelector) nominatePair(pair *CandidatePair) { } func (s *controllingSelector) HandleBindingRequest(message *stun.Message, local, remote Candidate) { //nolint:cyclop - reportPiggybacking(s.agent, message, remote) + s.agent.reportPiggybackingFromMessage(message, remote) s.agent.sendBindingSuccess(message, local, remote) @@ -183,7 +168,7 @@ func (s *controllingSelector) HandleSuccessResponse(message *stun.Message, local return } - reportPiggybacking(s.agent, message, remote) + s.agent.reportPiggybackingFromMessage(message, remote) s.log.Tracef("Inbound STUN (SuccessResponse) from %s to %s", remote, local) pair := s.agent.findPair(local, remote) @@ -224,14 +209,7 @@ func (s *controllingSelector) PingCandidate(local, remote Candidate) { AttrControlling(s.agent.tieBreaker), PriorityAttr(local.Priority()), } - if packet, acks := s.agent.GetPiggybackDataAndAcks(); acks != nil { - if acks != nil { - attributes = append(attributes, DtlsInStunAckAttribute(acks)) - } - if packet != nil { - attributes = append(attributes, DtlsInStunAttribute(packet)) - } - } + attributes = s.agent.appendPiggybackAttributes(attributes) attributes = append(attributes, stun.NewShortTermIntegrity(s.agent.remotePwd), stun.Fingerprint) @@ -384,14 +362,7 @@ func (s *controlledSelector) PingCandidate(local, remote Candidate) { AttrControlled(s.agent.tieBreaker), PriorityAttr(local.Priority()), } - if packet, acks := s.agent.GetPiggybackDataAndAcks(); acks != nil { - if acks != nil { - attributes = append(attributes, DtlsInStunAckAttribute(acks)) - } - if packet != nil { - attributes = append(attributes, DtlsInStunAttribute(packet)) - } - } + attributes = s.agent.appendPiggybackAttributes(attributes) attributes = append(attributes, stun.NewShortTermIntegrity(s.agent.remotePwd), stun.Fingerprint) @@ -462,11 +433,11 @@ func (s *controlledSelector) HandleSuccessResponse(message *stun.Message, local, pair.UpdateRoundTripTime(rtt) - reportPiggybacking(s.agent, message, remote) + s.agent.reportPiggybackingFromMessage(message, remote) } func (s *controlledSelector) HandleBindingRequest(message *stun.Message, local, remote Candidate) { //nolint:cyclop - reportPiggybacking(s.agent, message, remote) + s.agent.reportPiggybackingFromMessage(message, remote) pair := s.agent.findPair(local, remote) if pair == nil {