<commit_before>// +build !js

// Package webrtc implements the WebRTC 1.0 as defined in W3C WebRTC specification document.
package webrtc

import (
	"crypto/ecdsa"
	"crypto/elliptic"
	"crypto/rand"
	"fmt"
	"io"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/pion/ice/v2"
	"github.com/pion/logging"
	"github.com/pion/rtcp"
	"github.com/pion/sdp/v3"
	"github.com/pion/webrtc/v3/internal/util"
	"github.com/pion/webrtc/v3/pkg/rtcerr"
)

// PeerConnection represents a WebRTC connection that establishes a
// peer-to-peer communications with another PeerConnection instance in a
// browser, or to another endpoint implementing the required protocols.
type PeerConnection struct {
	statsID string
	mu      sync.RWMutex

	// ops is an operations queue which will ensure the enqueued actions are
	// executed in order. It is used for asynchronously, but serially processing
	// remote and local descriptions
	ops *operations

	configuration Configuration

	currentLocalDescription  *SessionDescription
	pendingLocalDescription  *SessionDescription
	currentRemoteDescription *SessionDescription
	pendingRemoteDescription *SessionDescription
	signalingState           SignalingState
	iceConnectionState       ICEConnectionState
	connectionState          PeerConnectionState

	idpLoginURL *string

	isClosed               *atomicBool
	isNegotiationNeeded    *atomicBool
	negotiationNeededState negotiationNeededState

	lastOffer  string
	lastAnswer string

	// a value containing the last known greater mid value
	// we internally generate mids as numbers. Needed since JSEP
	// requires that when reusing a media section a new unique mid
	// should be defined (see JSEP 3.4.1).
	greaterMid int

	rtpTransceivers []*RTPTransceiver

	onSignalingStateChangeHandler     func(SignalingState)
	onICEConnectionStateChangeHandler func(ICEConnectionState)
	onConnectionStateChangeHandler    func(PeerConnectionState)
	onTrackHandler                    func(*Track, *RTPReceiver)
	onDataChannelHandler              func(*DataChannel)
	onNegotiationNeededHandler        func()

	iceGatherer   *ICEGatherer
	iceTransport  *ICETransport
	dtlsTransport *DTLSTransport
	sctpTransport *SCTPTransport

	// A reference to the associated API state used by this connection
	api *API
	log logging.LeveledLogger
}

// NewPeerConnection creates a peerconnection with the default
// codecs. See API.NewPeerConnection for details.
func NewPeerConnection(configuration Configuration) (*PeerConnection, error) {
	m := MediaEngine{}
	m.RegisterDefaultCodecs()
	api := NewAPI(WithMediaEngine(m))
	return api.NewPeerConnection(configuration)
}

// NewPeerConnection creates a new PeerConnection with the provided configuration against the received API object
func (api *API) NewPeerConnection(configuration Configuration) (*PeerConnection, error) {
	// https://w3c.github.io/webrtc-pc/#constructor (Step #2)
	// Some variables defined explicitly despite their implicit zero values to
	// allow better readability to understand what is happening.
	pc := &PeerConnection{
		statsID: fmt.Sprintf("PeerConnection-%d", time.Now().UnixNano()),
		configuration: Configuration{
			ICEServers:           []ICEServer{},
			ICETransportPolicy:   ICETransportPolicyAll,
			BundlePolicy:         BundlePolicyBalanced,
			RTCPMuxPolicy:        RTCPMuxPolicyRequire,
			Certificates:         []Certificate{},
			ICECandidatePoolSize: 0,
		},
		ops:                    newOperations(),
		isClosed:               &atomicBool{},
		isNegotiationNeeded:    &atomicBool{},
		negotiationNeededState: negotiationNeededStateEmpty,
		lastOffer:              "",
		lastAnswer:             "",
		greaterMid:             -1,
		signalingState:         SignalingStateStable,
		iceConnectionState:     ICEConnectionStateNew,
		connectionState:        PeerConnectionStateNew,

		api: api,
		log: api.settingEngine.LoggerFactory.NewLogger("pc"),
	}

	var err error
	if err = pc.initConfiguration(configuration); err != nil {
		return nil, err
	}

	pc.iceGatherer, err = pc.createICEGatherer()
	if err != nil {
		return nil, err
	}

	// Create the ice transport
	iceTransport := pc.createICETransport()
	pc.iceTransport = iceTransport

	// Create the DTLS transport
	dtlsTransport, err := pc.api.NewDTLSTransport(pc.iceTransport, pc.configuration.Certificates)
	if err != nil {
		return nil, err
	}
	pc.dtlsTransport = dtlsTransport

	// Create the SCTP transport
	pc.sctpTransport = pc.api.NewSCTPTransport(pc.dtlsTransport)

	// Wire up the on datachannel handler
	pc.sctpTransport.OnDataChannel(func(d *DataChannel) {
		pc.mu.RLock()
		handler := pc.onDataChannelHandler
		pc.mu.RUnlock()
		if handler != nil {
			handler(d)
		}
	})

	return pc, nil
}

// initConfiguration defines validation of the specified Configuration and
// its assignment to the internal configuration variable. This function differs
// from its SetConfiguration counterpart because most of the checks do not
// include verification statements related to the existing state. Thus the
// function describes only minor verification of some the struct variables.
func (pc *PeerConnection) initConfiguration(configuration Configuration) error {
	if configuration.PeerIdentity != "" {
		pc.configuration.PeerIdentity = configuration.PeerIdentity
	}

	// https://www.w3.org/TR/webrtc/#constructor (step #3)
	if len(configuration.Certificates) > 0 {
		now := time.Now()
		for _, x509Cert := range configuration.Certificates {
			if !x509Cert.Expires().IsZero() && now.After(x509Cert.Expires()) {
				return &rtcerr.InvalidAccessError{Err: ErrCertificateExpired}
			}
			pc.configuration.Certificates = append(pc.configuration.Certificates, x509Cert)
		}
	} else {
		sk, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
		if err != nil {
			return &rtcerr.UnknownError{Err: err}
		}
		certificate, err := GenerateCertificate(sk)
		if err != nil {
			return err
		}
		pc.configuration.Certificates = []Certificate{*certificate}
	}

	if configuration.BundlePolicy != BundlePolicy(Unknown) {
		pc.configuration.BundlePolicy = configuration.BundlePolicy
	}

	if configuration.RTCPMuxPolicy != RTCPMuxPolicy(Unknown) {
		pc.configuration.RTCPMuxPolicy = configuration.RTCPMuxPolicy
	}

	if configuration.ICECandidatePoolSize != 0 {
		pc.configuration.ICECandidatePoolSize = configuration.ICECandidatePoolSize
	}

	if configuration.ICETransportPolicy != ICETransportPolicy(Unknown) {
		pc.configuration.ICETransportPolicy = configuration.ICETransportPolicy
	}

	if configuration.SDPSemantics != SDPSemantics(Unknown) {
		pc.configuration.SDPSemantics = configuration.SDPSemantics
	}

	sanitizedICEServers := configuration.getICEServers()
	if len(sanitizedICEServers) > 0 {
		for _, server := range sanitizedICEServers {
			if err := server.validate(); err != nil {
				return err
			}
		}
		pc.configuration.ICEServers = sanitizedICEServers
	}

	return nil
}

// OnSignalingStateChange sets an event handler which is invoked when the
// peer connection's signaling state changes
func (pc *PeerConnection) OnSignalingStateChange(f func(SignalingState)) {
	pc.mu.Lock()
	defer pc.mu.Unlock()
	pc.onSignalingStateChangeHandler = f
}

func (pc *PeerConnection) onSignalingStateChange(newState SignalingState) {
	pc.mu.RLock()
	handler := pc.onSignalingStateChangeHandler
	pc.mu.RUnlock()

	pc.log.Infof("signaling state changed to %s", newState)
	if handler != nil {
		go handler(newState)
	}
}

// OnDataChannel sets an event handler which is invoked when a data
// channel message arrives from a remote peer.
func (pc *PeerConnection) OnDataChannel(f func(*DataChannel)) {
	pc.mu.Lock()
	defer pc.mu.Unlock()
	pc.onDataChannelHandler = f
}

// OnNegotiationNeeded sets an event handler which is invoked when
// a change has occurred which requires session negotiation
func (pc *PeerConnection) OnNegotiationNeeded(f func()) {
	pc.mu.Lock()
	defer pc.mu.Unlock()
	pc.onNegotiationNeededHandler = f
}

func (pc *PeerConnection) onNegotiationNeeded() {
	// https://w3c.github.io/webrtc-pc/#updating-the-negotiation-needed-flag
	// non-canon step 1
	pc.mu.Lock()
	defer pc.mu.Unlock()
	if pc.negotiationNeededState == negotiationNeededStateRun {
		pc.negotiationNeededState = negotiationNeededStateQueue
		return
	} else if pc.negotiationNeededState == negotiationNeededStateQueue {
		return
	}

	pc.negotiationNeededState = negotiationNeededStateRun

	pc.ops.Enqueue(pc.negotiationNeededOp)
}

func (pc *PeerConnection) negotiationNeededOp() {
	// https://www.w3.org/TR/webrtc/#updating-the-negotiation-needed-flag
	// Step 2.1
	if pc.isClosed.get() {
		return
	}
	// non-canon step 2.2
	if !pc.ops.IsEmpty() {
		pc.ops.Enqueue(pc.negotiationNeededOp)
		return
	}

	// non-canon, run again if there was a request
	defer func() {
		pc.mu.Lock()
		if pc.negotiationNeededState == negotiationNeededStateQueue {
			defer pc.onNegotiationNeeded()
		}
		pc.negotiationNeededState = negotiationNeededStateEmpty
		pc.mu.Unlock()
	}()

	// Step 2.3
	if pc.SignalingState() != SignalingStateStable {
		return
	}

	// Step 2.4
	if !pc.checkNegotiationNeeded() {
		pc.isNegotiationNeeded.set(false)
		return
	}

	// Step 2.5
	if pc.isNegotiationNeeded.get() {
		return
	}

	// Step 2.6
	pc.isNegotiationNeeded.set(true)

	// Step 2.7
	if pc.onNegotiationNeededHandler != nil {
		pc.onNegotiationNeededHandler()
	}
}

func (pc *PeerConnection) checkNegotiationNeeded() bool { //nolint:gocognit
	// To check if negotiation is needed for connection, perform the following checks:
	// Skip 1, 2 steps
	// Step 3
	pc.mu.RLock()
	localDesc := pc.currentLocalDescription
	remoteDesc := pc.currentRemoteDescription
	pc.mu.RUnlock()

	if localDesc == nil {
		return true
	}

	pc.sctpTransport.lock.Lock()
	lenDataChannel := len(pc.sctpTransport.dataChannels)
	pc.sctpTransport.lock.Unlock()

	if lenDataChannel != 0 && haveDataChannel(localDesc) == nil {
		return true
	}

	for _, t := range pc.GetTransceivers() {
		// https://www.w3.org/TR/webrtc/#dfn-update-the-negotiation-needed-flag
		// Step 5.1
		// if t.stopping && !t.stopped {
		// 	return true
		// }
		m := getByMid(t.Mid(), localDesc)
		// Step 5.2
		if !t.stopped && m == nil {
			return true
		}
		if !t.stopped && m != nil {
			// Step 5.3.1
			if t.Direction() == RTPTransceiverDirectionSendrecv || t.Direction() == RTPTransceiverDirectionSendonly {
				descMsid, okMsid := m.Attribute(sdp.AttrKeyMsid)
				if !okMsid || descMsid != t.Sender().Track().Msid() {
					return true
				}
			}
			switch localDesc.Type {
			case SDPTypeOffer:<commit_msg>hold lock for checkNegotiationNeeded transceiver checks<commit_after>// +build !js

// Package webrtc implements the WebRTC 1.0 as defined in W3C WebRTC specification document.
package webrtc

import (
	"crypto/ecdsa"
	"crypto/elliptic"
	"crypto/rand"
	"fmt"
	"io"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/pion/ice/v2"
	"github.com/pion/logging"
	"github.com/pion/rtcp"
	"github.com/pion/sdp/v3"
	"github.com/pion/webrtc/v3/internal/util"
	"github.com/pion/webrtc/v3/pkg/rtcerr"
)

// PeerConnection represents a WebRTC connection that establishes a
// peer-to-peer communications with another PeerConnection instance in a
// browser, or to another endpoint implementing the required protocols.
type PeerConnection struct {
	statsID string
	mu      sync.RWMutex

	// ops is an operations queue which will ensure the enqueued actions are
	// executed in order. It is used for asynchronously, but serially processing
	// remote and local descriptions
	ops *operations

	configuration Configuration

	currentLocalDescription  *SessionDescription
	pendingLocalDescription  *SessionDescription
	currentRemoteDescription *SessionDescription
	pendingRemoteDescription *SessionDescription
	signalingState           SignalingState
	iceConnectionState       ICEConnectionState
	connectionState          PeerConnectionState

	idpLoginURL *string

	isClosed               *atomicBool
	isNegotiationNeeded    *atomicBool
	negotiationNeededState negotiationNeededState

	lastOffer  string
	lastAnswer string

	// a value containing the last known greater mid value
	// we internally generate mids as numbers. Needed since JSEP
	// requires that when reusing a media section a new unique mid
	// should be defined (see JSEP 3.4.1).
	greaterMid int

	rtpTransceivers []*RTPTransceiver

	onSignalingStateChangeHandler     func(SignalingState)
	onICEConnectionStateChangeHandler func(ICEConnectionState)
	onConnectionStateChangeHandler    func(PeerConnectionState)
	onTrackHandler                    func(*Track, *RTPReceiver)
	onDataChannelHandler              func(*DataChannel)
	onNegotiationNeededHandler        func()

	iceGatherer   *ICEGatherer
	iceTransport  *ICETransport
	dtlsTransport *DTLSTransport
	sctpTransport *SCTPTransport

	// A reference to the associated API state used by this connection
	api *API
	log logging.LeveledLogger
}

// NewPeerConnection creates a peerconnection with the default
// codecs. See API.NewPeerConnection for details.
func NewPeerConnection(configuration Configuration) (*PeerConnection, error) {
	m := MediaEngine{}
	m.RegisterDefaultCodecs()
	api := NewAPI(WithMediaEngine(m))
	return api.NewPeerConnection(configuration)
}

// NewPeerConnection creates a new PeerConnection with the provided configuration against the received API object
func (api *API) NewPeerConnection(configuration Configuration) (*PeerConnection, error) {
	// https://w3c.github.io/webrtc-pc/#constructor (Step #2)
	// Some variables defined explicitly despite their implicit zero values to
	// allow better readability to understand what is happening.
	pc := &PeerConnection{
		statsID: fmt.Sprintf("PeerConnection-%d", time.Now().UnixNano()),
		configuration: Configuration{
			ICEServers:           []ICEServer{},
			ICETransportPolicy:   ICETransportPolicyAll,
			BundlePolicy:         BundlePolicyBalanced,
			RTCPMuxPolicy:        RTCPMuxPolicyRequire,
			Certificates:         []Certificate{},
			ICECandidatePoolSize: 0,
		},
		ops:                    newOperations(),
		isClosed:               &atomicBool{},
		isNegotiationNeeded:    &atomicBool{},
		negotiationNeededState: negotiationNeededStateEmpty,
		lastOffer:              "",
		lastAnswer:             "",
		greaterMid:             -1,
		signalingState:         SignalingStateStable,
		iceConnectionState:     ICEConnectionStateNew,
		connectionState:        PeerConnectionStateNew,

		api: api,
		log: api.settingEngine.LoggerFactory.NewLogger("pc"),
	}

	var err error
	if err = pc.initConfiguration(configuration); err != nil {
		return nil, err
	}

	pc.iceGatherer, err = pc.createICEGatherer()
	if err != nil {
		return nil, err
	}

	// Create the ice transport
	iceTransport := pc.createICETransport()
	pc.iceTransport = iceTransport

	// Create the DTLS transport
	dtlsTransport, err := pc.api.NewDTLSTransport(pc.iceTransport, pc.configuration.Certificates)
	if err != nil {
		return nil, err
	}
	pc.dtlsTransport = dtlsTransport

	// Create the SCTP transport
	pc.sctpTransport = pc.api.NewSCTPTransport(pc.dtlsTransport)

	// Wire up the on datachannel handler
	pc.sctpTransport.OnDataChannel(func(d *DataChannel) {
		pc.mu.RLock()
		handler := pc.onDataChannelHandler
		pc.mu.RUnlock()
		if handler != nil {
			handler(d)
		}
	})

	return pc, nil
}

// initConfiguration defines validation of the specified Configuration and
// its assignment to the internal configuration variable. This function differs
// from its SetConfiguration counterpart because most of the checks do not
// include verification statements related to the existing state. Thus the
// function describes only minor verification of some the struct variables.
func (pc *PeerConnection) initConfiguration(configuration Configuration) error {
	if configuration.PeerIdentity != "" {
		pc.configuration.PeerIdentity = configuration.PeerIdentity
	}

	// https://www.w3.org/TR/webrtc/#constructor (step #3)
	if len(configuration.Certificates) > 0 {
		now := time.Now()
		for _, x509Cert := range configuration.Certificates {
			if !x509Cert.Expires().IsZero() && now.After(x509Cert.Expires()) {
				return &rtcerr.InvalidAccessError{Err: ErrCertificateExpired}
			}
			pc.configuration.Certificates = append(pc.configuration.Certificates, x509Cert)
		}
	} else {
		sk, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader)
		if err != nil {
			return &rtcerr.UnknownError{Err: err}
		}
		certificate, err := GenerateCertificate(sk)
		if err != nil {
			return err
		}
		pc.configuration.Certificates = []Certificate{*certificate}
	}

	if configuration.BundlePolicy != BundlePolicy(Unknown) {
		pc.configuration.BundlePolicy = configuration.BundlePolicy
	}

	if configuration.RTCPMuxPolicy != RTCPMuxPolicy(Unknown) {
		pc.configuration.RTCPMuxPolicy = configuration.RTCPMuxPolicy
	}

	if configuration.ICECandidatePoolSize != 0 {
		pc.configuration.ICECandidatePoolSize = configuration.ICECandidatePoolSize
	}

	if configuration.ICETransportPolicy != ICETransportPolicy(Unknown) {
		pc.configuration.ICETransportPolicy = configuration.ICETransportPolicy
	}

	if configuration.SDPSemantics != SDPSemantics(Unknown) {
		pc.configuration.SDPSemantics = configuration.SDPSemantics
	}

	sanitizedICEServers := configuration.getICEServers()
	if len(sanitizedICEServers) > 0 {
		for _, server := range sanitizedICEServers {
			if err := server.validate(); err != nil {
				return err
			}
		}
		pc.configuration.ICEServers = sanitizedICEServers
	}

	return nil
}

// OnSignalingStateChange sets an event handler which is invoked when the
// peer connection's signaling state changes
func (pc *PeerConnection) OnSignalingStateChange(f func(SignalingState)) {
	pc.mu.Lock()
	defer pc.mu.Unlock()
	pc.onSignalingStateChangeHandler = f
}

func (pc *PeerConnection) onSignalingStateChange(newState SignalingState) {
	pc.mu.RLock()
	handler := pc.onSignalingStateChangeHandler
	pc.mu.RUnlock()

	pc.log.Infof("signaling state changed to %s", newState)
	if handler != nil {
		go handler(newState)
	}
}

// OnDataChannel sets an event handler which is invoked when a data
// channel message arrives from a remote peer.
func (pc *PeerConnection) OnDataChannel(f func(*DataChannel)) {
	pc.mu.Lock()
	defer pc.mu.Unlock()
	pc.onDataChannelHandler = f
}

// OnNegotiationNeeded sets an event handler which is invoked when
// a change has occurred which requires session negotiation
func (pc *PeerConnection) OnNegotiationNeeded(f func()) {
	pc.mu.Lock()
	defer pc.mu.Unlock()
	pc.onNegotiationNeededHandler = f
}

func (pc *PeerConnection) onNegotiationNeeded() {
	// https://w3c.github.io/webrtc-pc/#updating-the-negotiation-needed-flag
	// non-canon step 1
	pc.mu.Lock()
	defer pc.mu.Unlock()
	if pc.negotiationNeededState == negotiationNeededStateRun {
		pc.negotiationNeededState = negotiationNeededStateQueue
		return
	} else if pc.negotiationNeededState == negotiationNeededStateQueue {
		return
	}

	pc.negotiationNeededState = negotiationNeededStateRun

	pc.ops.Enqueue(pc.negotiationNeededOp)
}

func (pc *PeerConnection) negotiationNeededOp() {
	// https://www.w3.org/TR/webrtc/#updating-the-negotiation-needed-flag
	// Step 2.1
	if pc.isClosed.get() {
		return
	}
	// non-canon step 2.2
	if !pc.ops.IsEmpty() {
		pc.ops.Enqueue(pc.negotiationNeededOp)
		return
	}

	// non-canon, run again if there was a request
	defer func() {
		pc.mu.Lock()
		if pc.negotiationNeededState == negotiationNeededStateQueue {
			defer pc.onNegotiationNeeded()
		}
		pc.negotiationNeededState = negotiationNeededStateEmpty
		pc.mu.Unlock()
	}()

	// Step 2.3
	if pc.SignalingState() != SignalingStateStable {
		return
	}

	// Step 2.4
	if !pc.checkNegotiationNeeded() {
		pc.isNegotiationNeeded.set(false)
		return
	}

	// Step 2.5
	if pc.isNegotiationNeeded.get() {
		return
	}

	// Step 2.6
	pc.isNegotiationNeeded.set(true)

	// Step 2.7
	if pc.onNegotiationNeededHandler != nil {
		pc.onNegotiationNeededHandler()
	}
}

func (pc *PeerConnection) checkNegotiationNeeded() bool { //nolint:gocognit
	// To check if negotiation is needed for connection, perform the following checks:
	// Skip 1, 2 steps
	// Step 3
	pc.mu.RLock()
	localDesc := pc.currentLocalDescription
	remoteDesc := pc.currentRemoteDescription
	pc.mu.RUnlock()

	if localDesc == nil {
		return true
	}

	pc.sctpTransport.lock.Lock()
	lenDataChannel := len(pc.sctpTransport.dataChannels)
	pc.sctpTransport.lock.Unlock()

	if lenDataChannel != 0 && haveDataChannel(localDesc) == nil {
		return true
	}

	pc.mu.Lock()
	defer pc.mu.Unlock()

	for _, t := range pc.rtpTransceivers {
		// https://www.w3.org/TR/webrtc/#dfn-update-the-negotiation-needed-flag
		// Step 5.1
		// if t.stopping && !t.stopped {
		// 	return true
		// }
		m := getByMid(t.Mid(), localDesc)
		// Step 5.2
		if !t.stopped && m == nil {
			return true
		}
		if !t.stopped && m != nil {
			// Step 5.3.1
			if t.Direction() == RTPTransceiverDirectionSendrecv || t.Direction() == RTPTransceiverDirectionSendonly {
				descMsid, okMsid := m.Attribute(sdp.AttrKeyMsid)
				if !okMsid || descMsid != t.Sender().Track().Msid() {
					return true
				}
			}
			switch localDesc.Type {
			case SDPTypeOffer:
{'fraction_non_alphanumeric': 0.09961540287735625, 'fraction_numerical': 0.005982621907791653, 'mean_word_length': 3.988630980577925, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 7, 'https://': 10, 'lorem ipsum': 0, 'www.': 6, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1021925', 'n_tokens_mistral': 7078, 'n_tokens_neox': 6546, 'n_words': 1607}