diff --git a/agent.go b/agent.go index 5350d1d2..e14fa12d 100644 --- a/agent.go +++ b/agent.go @@ -178,6 +178,8 @@ type Agent struct { lastRenominationTime time.Time turnClientFactory func(*turn.ClientConfig) (turnClient, error) + + piggyback piggybackingController } // NewAgent creates a new Agent. @@ -221,6 +223,10 @@ 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.init() + return newAgentWithConfig(agent, opts...) } @@ -744,21 +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() - } + if a.connectionState == newState { + return + } - a.log.Infof("Setting new connection state: %s", newState) - a.connectionState = newState - a.connectionStateNotifier.EnqueueConnectionState(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() + } + + 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) } func (a *Agent) setSelectedPair(pair *CandidatePair) { @@ -1582,6 +1602,7 @@ func (a *Agent) sendBindingSuccess(m *stun.Message, local, remote Candidate) { Port: port, }, } + attributes = a.appendPiggybackAttributes(attributes) attributes = append(attributes, stun.NewShortTermIntegrity(a.localPwd), stun.Fingerprint) @@ -2053,6 +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) } + attributes = a.appendPiggybackAttributes(attributes) attributes = append(attributes, stun.NewShortTermIntegrity(a.remotePwd), diff --git a/piggyback.go b/piggyback.go new file mode 100644 index 00000000..4bb349a5 --- /dev/null +++ b/piggyback.go @@ -0,0 +1,236 @@ +// 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" + + "github.com/pion/stun/v3" +) + +type packetWithCrc struct { + data []byte + crc uint32 +} + +type piggybackingState int + +const ( + PiggybackingStateTentative piggybackingState = 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 +} + +// 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. +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) + } +} + +// 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() + + 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) + }) +} diff --git a/selection.go b/selection.go index 0ebd8ced..83624c44 100644 --- a/selection.go +++ b/selection.go @@ -92,6 +92,7 @@ func (s *controllingSelector) nominatePair(pair *CandidatePair) { AttrControlling(s.agent.tieBreaker), PriorityAttr(pair.Local.Priority()), } + attributes = s.agent.appendPiggybackAttributes(attributes) attributes = append(attributes, stun.NewShortTermIntegrity(s.agent.remotePwd), stun.Fingerprint) @@ -107,6 +108,8 @@ func (s *controllingSelector) nominatePair(pair *CandidatePair) { } func (s *controllingSelector) HandleBindingRequest(message *stun.Message, local, remote Candidate) { //nolint:cyclop + s.agent.reportPiggybackingFromMessage(message, remote) + s.agent.sendBindingSuccess(message, local, remote) pair := s.agent.findPair(local, remote) @@ -141,10 +144,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 +168,8 @@ func (s *controllingSelector) HandleSuccessResponse(m *stun.Message, local, remo return } + s.agent.reportPiggybackingFromMessage(message, remote) + s.log.Tracef("Inbound STUN (SuccessResponse) from %s to %s", remote, local) pair := s.agent.findPair(local, remote) @@ -202,6 +209,7 @@ func (s *controllingSelector) PingCandidate(local, remote Candidate) { AttrControlling(s.agent.tieBreaker), PriorityAttr(local.Priority()), } + attributes = s.agent.appendPiggybackAttributes(attributes) attributes = append(attributes, stun.NewShortTermIntegrity(s.agent.remotePwd), stun.Fingerprint) @@ -354,6 +362,7 @@ func (s *controlledSelector) PingCandidate(local, remote Candidate) { AttrControlled(s.agent.tieBreaker), PriorityAttr(local.Priority()), } + attributes = s.agent.appendPiggybackAttributes(attributes) attributes = append(attributes, stun.NewShortTermIntegrity(s.agent.remotePwd), stun.Fingerprint) @@ -368,7 +377,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 +388,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 +432,13 @@ func (s *controlledSelector) HandleSuccessResponse(m *stun.Message, local, remot } pair.UpdateRoundTripTime(rtt) + + s.agent.reportPiggybackingFromMessage(message, remote) } func (s *controlledSelector) HandleBindingRequest(message *stun.Message, local, remote Candidate) { //nolint:cyclop + s.agent.reportPiggybackingFromMessage(message, remote) + pair := s.agent.findPair(local, remote) if pair == nil { pair = s.agent.addPair(local, remote)