import { randomBytes } from "crypto" ; import { Uint64BE } from "int64-buffer" ; import * as nodeIp from "ip" ; import { isEqual , range } from "lodash" ; import { isIPv4 } from "net" ; import PCancelable from "p-cancelable" ; import { Event } from "rx.mini" ; import { Candidate , candidateFoundation , candidatePriority } from "./candidate" ; import { TransactionError } from "./exceptions" ; import { Address , Protocol } from "./types/model" ; import { createTurnEndpoint } from "./turn/protocol" ; import { difference , Future , future , PQueue , randomString , sleep , } from "./utils" ; import { StunProtocol } from "./stun/protocol" ; import { Message , parseMessage } from "./stun/message" ; import { classes , methods } from "./stun/const" ; import dns from "dns" ; import util from "util" ; const ICE_COMPLETED = 1 ; const ICE_FAILED = 2 ; const CONSENT_FAILURES = 6 ; const CONSENT_INTERVAL = 5 ; export enum CandidatePairState { FROZEN = 0 , WAITING = 1 , IN_PROGRESS = 2 , SUCCEEDED = 3 , FAILED = 4 } type IceState = "disconnected" | "closed" | "completed" ; export type IceOptions = { components : number ; stunServer ? : Address ; turnServer ? : Address ; turnUsername ? : string ; turnPassword ? : string ; turnSsl ? : boolean ; turnTransport ? : string ; forceTurn ? : boolean ; useIpv4 : boolean ; useIpv6 : boolean ; log : boolean ; } ; const defaultOptions = { components : 1 , useIpv4 : true , useIpv6 : true , log : false , } ; export class Connection { remotePassword : string = "" ; remoteUsername : string = "" ; localUserName = randomString ( 4 ) ; localPassword = randomString ( 22 ) ; remoteIsLite = false ; checkList : CandidatePair [ ] = [ ] ; localCandidates : Candidate [ ] = [ ] ; stunServer ? : Address ; useIpv4 : boolean ; useIpv6 : boolean ; options : IceOptions ; remoteCandidatesEnd = false ; _components : Set < number > ; _localCandidatesEnd = false ; _tieBreaker : BigInt = BigInt ( new Uint64BE ( randomBytes ( 64 ) ) . toString ( ) ) ; readonly onData = new Event < [ Buffer , number ] > ( ) ; readonly stateChanged = new Event < [ IceState ] > ( ) ; private _remoteCandidates : Candidate [ ] = [ ] ; private nominated : { [ key ] : CandidatePair ; } = { } ; get nominatedKeys ( ) { return Object . keys ( this . nominated ) . map ( ( v ) => v . toString ( ) ) ; } private nominating = new Set < number > ( ) ; get remoteAddr ( ) { return Object . values ( this . nominated ) [ 0 ] . remoteAddr ; } private checkListDone = false ; private checkListState = new PQueue < number > ( ) ; private earlyChecks : [ Message , Address , Protocol ] [ ] = [ ] ; private localCandidatesStart = false ; private protocols : Protocol [ ] = [ ] ; private queryConsentHandle ? : Future ; private promiseGatherCandidates ? : Event < [ ] > ; constructor ( public iceControlling , options ? ) { this . options = { ... defaultOptions , ... options , } ; const { components , stunServer , useIpv4 , useIpv6 } = this . options ; this . stunServer = stunServer ; this . useIpv4 = useIpv4 ; this . useIpv6 = useIpv6 ; this . _components = new Set ( range ( 1 , components + 1 ) ) ; } async gatherCandidates ( cb ? ) { if ( ! this . localCandidatesStart ) { this . localCandidatesStart = true ; this . promiseGatherCandidates = new Event ( ) ; const address = getHostAddress ( this . useIpv4 , this . useIpv6 ) ; for ( const component of this . _components ) { const candidates = await this . getComponentCandidates ( component , address , 5 , cb ) ; this . localCandidates = [ ... this . localCandidates , ... candidates ] ; } this . _localCandidatesEnd = true ; this . promiseGatherCandidates . execute ( ) ; } } private async getComponentCandidates ( component , addresses , timeout = 5 , cb ? ) { let candidates = [ ] ; for ( const address of addresses ) { const protocol = new StunProtocol ( this ) ; await protocol . connectionMade ( isIPv4 ( address ) ) ; protocol . localAddress = address ; this . protocols . push ( protocol ) ; const candidateAddress = protocol . getExtraInfo ; protocol . localCandidate = new Candidate ( candidateFoundation ( "host" , "udp" , candidateAddress [ 0 ] ) , component , "udp" , candidatePriority ( component , "host" ) , candidateAddress [ 0 ] , candidateAddress [ 1 ] , "host" ) ; candidates . push ( protocol . localCandidate ) ; if ( cb ) cb ( protocol . localCandidate ) ; } const stunServer = this . stunServer ; if ( stunServer ) { try { const srflxCandidates = ( await Promise . all < Candidate | void > ( this . protocols . map ( ( protocol ) => new Promise ( async ( r , f ) => { setTimeout ( f , timeout * 1000 ) ; if ( protocol . localCandidate ?. host && isIPv4 ( protocol . localCandidate ?. host ) ) { const candidate = await serverReflexiveCandidate ( protocol , stunServer ) . catch ( console . log ) ; if ( candidate && cb ) cb ( candidate ) ; r ( candidate ) ; } else { r ( ) ; } } ) ) ) ) . filter ( ( v ) => v ) as Candidate [ ] ; candidates = [ ... candidates , ... srflxCandidates ] ; } catch ( error ) { this . log ( "query STUN server" , error ) ; } } if ( this . options . turnServer && this . options . turnUsername && this . options . turnPassword ) { const protocol = await createTurnEndpoint ( this . options . turnServer , this . options . turnUsername , this . options . turnPassword ) ; this . protocols . push ( protocol ) ; const candidateAddress = protocol . turn . relayedAddress ; const relatedAddress = protocol . turn . mappedAddress ; protocol . localCandidate = new Candidate ( candidateFoundation ( "relay" , "udp" , candidateAddress [ 0 ] ) , component , "udp" , candidatePriority ( component , "relay" ) , candidateAddress [ 0 ] , candidateAddress [ 1 ] , "relay" , relatedAddress [ 0 ] , relatedAddress [ 1 ] ) ; protocol . receiver = this ; if ( this . options . forceTurn ) { candidates = [ ] ; } candidates . push ( protocol . localCandidate ) ; } return candidates ; } async connect ( ) { if ( ! this . _localCandidatesEnd ) { if ( ! this . localCandidatesStart ) throw new Error ( "Local candidates gathering was not performed" ) ; if ( this . promiseGatherCandidates ) await this . promiseGatherCandidates . asPromise ( ) ; } if ( ! this . remoteUsername || ! this . remotePassword ) throw new Error ( "Remote username or password is missing" ) ; this . remoteCandidates . forEach ( this . pairRemoteCandidate ) ; this . sortCheckList ( ) ; this . unfreezeInitial ( ) ; this . earlyChecks . forEach ( ( earlyCheck ) => this . checkIncoming ( ... earlyCheck ) ) ; this . earlyChecks = [ ] ; for ( ; ; ) { if ( ! this . schedulingChecks ( ) ) break ; await sleep ( 20 ) ; } const res = this . checkList . length > 0 ? await this . checkListState . get ( ) : ICE_FAILED ; this . checkList . forEach ( ( check ) => check . handle ?. cancel ( ) ) ; if ( res !== ICE_COMPLETED ) { throw new Error ( "ICE negotiation failed" ) ; } this . queryConsentHandle = future ( this . queryConsent ( ) ) ; this . stateChanged . execute ( "completed" ) ; } private unfreezeInitial ( ) { const firstPair = this . checkList . find ( ( pair ) => pair . component === Math . min ( ... [ ... this . _components ] ) ) ; if ( ! firstPair ) return ; if ( firstPair . state === CandidatePairState . FROZEN ) { this . checkState ( firstPair , CandidatePairState . WAITING ) ; } const seenFoundations = new Set ( firstPair . localCandidate . foundation ) ; for ( const pair of this . checkList ) { if ( pair . component === firstPair . component && ! seenFoundations . has ( pair . localCandidate . foundation ) && pair . state === CandidatePairState . FROZEN ) { this . checkState ( pair , CandidatePairState . WAITING ) ; seenFoundations . add ( pair . localCandidate . foundation ) ; } } } private schedulingChecks ( ) { { const pair = this . checkList . filter ( ( pair ) => { if ( this . options . forceTurn && pair . protocol . type === "stun" ) return false ; return true ; } ) . find ( ( pair ) => pair . state === CandidatePairState . WAITING ) ; if ( pair ) { pair . handle = future ( this . checkStart ( pair ) ) ; return true ; } } { const pair = this . checkList . find ( ( pair ) => pair . state === CandidatePairState . FROZEN ) ; if ( pair ) { pair . handle = future ( this . checkStart ( pair ) ) ; return true ; } } if ( ! this . remoteCandidatesEnd ) { return ! this . checkListDone ; } return false ; } private queryConsent = ( ) => new PCancelable ( async ( r , f , onCancel ) => { let failures = 0 ; onCancel ( ( ) => { failures += CONSENT_FAILURES ; f ( "cancel" ) ; } ) ; for ( ; ; ) { await sleep ( CONSENT_INTERVAL * ( 0.8 + 0.4 * Math . random ( ) ) * 1000 ) ; for ( const key of this . nominatedKeys ) { const pair = this . nominated [ Number ( key ) ] ; const request = this . buildRequest ( pair , false ) ; try { await pair . protocol . request ( request , pair . remoteAddr , Buffer . from ( this . remotePassword , "utf8" ) , 0 ) ; failures = 0 ; } catch ( error ) { failures ++ ; this . stateChanged . execute ( "disconnected" ) ; } if ( failures >= CONSENT_FAILURES ) { this . log ( "Consent to send expired" ) ; this . queryConsentHandle = undefined ; r ( await this . close ( ) ) ; return ; } } } } ) ; async close ( ) { if ( this . queryConsentHandle && ! this . queryConsentHandle . done ( ) ) { this . queryConsentHandle . cancel ( ) ; try { await this . queryConsentHandle . promise ; } catch ( error ) { } } if ( this . checkList && ! this . checkListDone ) { this . checkListState . put ( new Promise ( ( r ) => r ( ICE_FAILED ) ) ) ; } this . nominated = { } ; for ( const protocol of this . protocols ) { if ( protocol . close ) await protocol . close ( ) ; } this . protocols = [ ] ; this . localCandidates = [ ] ; this . stateChanged . execute ( "closed" ) ; } async addRemoteCandidate ( remoteCandidate ) { if ( this . remoteCandidatesEnd ) throw new Error ( "Cannot add remote candidate after end-of-candidates." ) ; if ( ! remoteCandidate ) { this . pruneComponents ( ) ; this . remoteCandidatesEnd = true ; return ; } if ( remoteCandidate . host . includes ( ".local" ) ) { await sleep ( 10 ) ; const res = await util . promisify ( dns . lookup ) ( remoteCandidate . host ) . catch ( ( ) => { } ) ; if ( ! res ) return ; remoteCandidate . host = res . address ; } try { validateRemoteCandidate ( remoteCandidate ) ; } catch ( error ) { return ; } this . remoteCandidates . push ( remoteCandidate ) ; this . pairRemoteCandidate ( remoteCandidate ) ; this . sortCheckList ( ) ; } async send ( data ) { await this . sendTo ( data , 1 ) ; } private async sendTo ( data , component ) { const activePair = this . nominated [ component ] ; if ( activePair ) { await activePair . protocol . sendData ( data , activePair . remoteAddr ) ; } else { throw new Error ( "Cannot send data, not connected" ) ; } } getDefaultCandidate ( component ) { const candidates = this . localCandidates . sort ( ( a , b ) => a . priority - b . priority ) ; const candidate = candidates . find ( ( candidate ) => candidate . component === component ) ; return candidate ; } requestReceived ( message , addr , protocol , rawData ) { if ( message . messageMethod !== methods . BINDING ) { this . respondError ( message , addr , protocol , [ 400 , "Bad Request" ] ) ; return ; } try { parseMessage ( rawData , Buffer . from ( this . localPassword , "utf8" ) ) ; if ( ! this . remoteUsername ) { const rxUsername = `${ this . localUserName }:${ this . remoteUsername }` ; if ( message . attributes [ "USERNAME" ] != rxUsername ) throw new Error ( "Wrong username" ) ; } } catch ( error ) { this . respondError ( message , addr , protocol , [ 400 , "Bad Request" ] ) ; return ; } if ( this . iceControlling && message . attributesKeys . includes ( "ICE-CONTROLLING" ) ) { if ( this . _tieBreaker >= message . attributes [ "ICE-CONTROLLING" ] ) { this . respondError ( message , addr , protocol , [ 487 , "Role Conflict" ] ) ; return ; } else { this . switchRole ( false ) ; } } else if ( ! this . iceControlling && message . attributesKeys . includes ( "ICE-CONTROLLED" ) ) { if ( this . _tieBreaker < message . attributes [ "ICE-CONTROLLED" ] ) { this . respondError ( message , addr , protocol , [ 487 , "Role Conflict" ] ) ; } else { this . switchRole ( true ) ; return ; } } const response = new Message ( methods . BINDING , classes . RESPONSE , message . transactionId ) ; response . attributes [ "XOR-MAPPED-ADDRESS" ] = addr ; response . addMessageIntegrity ( Buffer . from ( this . localPassword , "utf8" ) ) ; response . addFingerprint ( ) ; protocol . sendStun ( response , addr ) ; if ( ! this . checkList ) { this . earlyChecks . push ( [ message , addr , protocol ] ) ; } else { this . checkIncoming ( message , addr , protocol ) ; } } dataReceived ( data , component ) { this . onData . execute ( data , component ) ; } private log ( ... args ) { if ( this . options . log ) { console . log ( "log" , ... args ) ; } } set remoteCandidates ( value ) { if ( this . remoteCandidatesEnd ) throw new Error ( "Cannot set remote candidates after end-of-candidates." ) ; this . _remoteCandidates = [ ] ; for ( const remoteCandidate of value ) { try { validateRemoteCandidate ( remoteCandidate ) ; } catch ( error ) { continue ; } this . remoteCandidates . push ( remoteCandidate ) ; } this . pruneComponents ( ) ; this . remoteCandidatesEnd = true ; } get remoteCandidates ( ) { return this . _remoteCandidates ; } private pruneComponents ( ) { const seenComponents = new Set ( this . remoteCandidates . map ( ( v ) => v . component ) ) ; const missingComponents = [ ... difference ( this . _components , seenComponents ) ] ; if ( missingComponents . length > 0 ) { this . _components = seenComponents ; } } private sortCheckList ( ) { sortCandidatePairs ( this . checkList , this . iceControlling ) ; } private findPair ( protocol , remoteCandidate ) { const pair = this . checkList . find ( ( pair ) => isEqual ( pair . protocol , protocol ) && isEqual ( pair . remoteCandidate , remoteCandidate ) ) ; return pair ; } private checkState ( pair , state ) { pair . state = state ; } private switchRole ( iceControlling ) { this . iceControlling = iceControlling ; this . sortCheckList ( ) ; } private checkComplete ( pair ) { pair . handle = undefined ; if ( pair . state === CandidatePairState . SUCCEEDED ) { if ( pair . nominated ) { this . nominated [ pair . component ] = pair ; for ( const p of this . checkList ) { if ( p . component === pair . component && [ CandidatePairState . WAITING , CandidatePairState . FROZEN ] . includes ( p . state ) ) { this . checkState ( p , CandidatePairState . FAILED ) ; } } } if ( this . nominatedKeys . length === this . _components . size ) { if ( ! this . checkListDone ) { this . log ( "ICE completed" ) ; this . checkListState . put ( new Promise ( ( r ) => r ( ICE_COMPLETED ) ) ) ; this . checkListDone = true ; } return ; } for ( const p of this . checkList ) { if ( p . localCandidate . foundation === pair . localCandidate . foundation && p . state === CandidatePairState . FROZEN ) { this . checkState ( p , CandidatePairState . WAITING ) ; } } } { const list = [ CandidatePairState . SUCCEEDED , CandidatePairState . FAILED ] ; if ( this . checkList . find ( ( { state } ) => ! list . includes ( state ) ) ) { return ; } } if ( ! this . iceControlling ) { const target = CandidatePairState . SUCCEEDED ; if ( this . checkList . find ( ( { state } ) => state === target ) ) { return ; } } if ( ! this . checkListDone ) { this . log ( "ICE failed" ) ; this . checkListState . put ( new Promise ( ( r ) => r ( ICE_FAILED ) ) ) ; this . checkListDone = true ; } } checkStart = ( pair ) => new PCancelable ( async ( r , f , onCancel ) => { onCancel ( ( ) => f ( "cancel" ) ) ; this . checkState ( pair , CandidatePairState . IN_PROGRESS ) ; const nominate = this . iceControlling && ! this . remoteIsLite ; const request = this . buildRequest ( pair , nominate ) ; const result = { } ; try { const [ response , addr ] = await pair . protocol . request ( request , pair . remoteAddr , Buffer . from ( this . remotePassword , "utf8" ) ) ; result . response = response ; result . addr = addr ; } catch ( error ) { const exc = error ; if ( exc . response ?. attributes [ "ERROR-CODE" ] [ 0 ] === 487 ) { if ( request . attributesKeys . includes ( "ICE-CONTROLLED" ) ) { this . switchRole ( true ) ; } else if ( request . attributesKeys . includes ( "ICE-CONTROLLING" ) ) { this . switchRole ( false ) ; } await this . checkStart ( pair ) ; r ( ) ; return ; } else { this . checkState ( pair , CandidatePairState . FAILED ) ; this . checkComplete ( pair ) ; r ( ) ; return ; } } if ( ! isEqual ( result . addr , pair . remoteAddr ) ) { this . checkState ( pair , CandidatePairState . FAILED ) ; this . checkComplete ( pair ) ; r ( ) ; return ; } if ( nominate || pair . remoteNominated ) { pair . nominated = true ; } else if ( this . iceControlling && ! this . nominating . has ( pair . component ) ) { this . nominating . add ( pair . component ) ; const request = this . buildRequest ( pair , true ) ; try { await pair . protocol . request ( request , pair . remoteAddr , Buffer . from ( this . remotePassword , "utf8" ) ) ; } catch ( error ) { this . checkState ( pair , CandidatePairState . FAILED ) ; this . checkComplete ( pair ) ; return ; } pair . nominated = true ; } this . checkState ( pair , CandidatePairState . SUCCEEDED ) ; this . checkComplete ( pair ) ; r ( ) ; } ) ; checkIncoming ( message , addr , protocol ) { const component = protocol . localCandidate ?. component ; if ( component == undefined ) throw new Error ( ) ; let remoteCandidate ; const [ host , port ] = addr ; for ( const c of this . remoteCandidates ) { if ( c . host === host && c . port === port ) { remoteCandidate = c ; if ( remoteCandidate . component !== component ) throw new Error ( "checkIncoming" ) ; break ; } } if ( ! remoteCandidate ) { remoteCandidate = new Candidate ( randomString ( 10 ) , component , "udp" , message . attributes [ "PRIORITY" ] , host , port , "prflx" ) ; this . remoteCandidates . push ( remoteCandidate ) ; } let pair = this . findPair ( protocol , remoteCandidate ) ; if ( ! pair ) { pair = new CandidatePair ( protocol , remoteCandidate ) ; pair . state = CandidatePairState . WAITING ; this . checkList . push ( pair ) ; this . sortCheckList ( ) ; } if ( [ CandidatePairState . WAITING , CandidatePairState . FAILED ] . includes ( pair . state ) ) { pair . handle = future ( this . checkStart ( pair ) ) ; } if ( message . attributesKeys . includes ( "USE-CANDIDATE" ) && ! this . iceControlling ) { pair . remoteNominated = true ; if ( pair . state === CandidatePairState . SUCCEEDED ) { pair . nominated = true ; this . checkComplete ( pair ) ; } } } private pairRemoteCandidate = ( remoteCandidate ) => { for ( const protocol of this . protocols ) { if ( protocol . localCandidate ?. canPairWith ( remoteCandidate ) && ! this . findPair ( protocol , remoteCandidate ) ) { const pair = new CandidatePair ( protocol , remoteCandidate ) ; this . checkList . push ( pair ) ; } } } ; private buildRequest ( pair , nominate ) { const txUsername = `${ this . remoteUsername }:${ this . localUserName }` ; const request = new Message ( methods . BINDING , classes . REQUEST ) ; request . attributes [ "USERNAME" ] = txUsername ; request . attributes [ "PRIORITY" ] = candidatePriority ( pair . component , "prflx" ) ; if ( this . iceControlling ) { request . attributes [ "ICE-CONTROLLING" ] = this . _tieBreaker ; if ( nominate ) { request . attributes [ "USE-CANDIDATE" ] = null ; } } else { request . attributes [ "ICE-CONTROLLED" ] = this . _tieBreaker ; } return request ; } private respondError ( request , addr , protocol , errorCode ) { const response = new Message ( request . messageMethod , classes . ERROR , request . transactionId ) ; response . attributes [ "ERROR-CODE" ] = errorCode ; response . addMessageIntegrity ( Buffer . from ( this . localPassword , "utf8" ) ) ; response . addFingerprint ( ) ; protocol . sendStun ( response , addr ) ; } } export class CandidatePair { handle ? : Future ; nominated = false ; remoteNominated = false ; state = CandidatePairState . FROZEN ; constructor ( public protocol , public remoteCandidate ) { } get localCandidate ( ) { if ( ! this . protocol . localCandidate ) throw new Error ( "localCandidate not exist" ) ; return this . protocol . localCandidate ; } get remoteAddr ( ) : Address { return [ this . remoteCandidate . host , this . remoteCandidate . port ] ; } get component ( ) { return this . localCandidate . component ; } } export function validateRemoteCandidate ( candidate ) { if ( ! [ "host" , "relay" , "srflx" ] . includes ( candidate . type ) ) throw new Error ( `Unexpected candidate type "${ candidate . type }"` ) ; return candidate ; } export function sortCandidatePairs ( pairs , iceControlling ) { pairs . sort ( ( a , b ) => candidatePairPriority ( a . localCandidate , a . remoteCandidate , iceControlling ) - candidatePairPriority ( b . localCandidate , b . remoteCandidate , iceControlling ) ) ; } export function candidatePairPriority ( local , remote , iceControlling ) { const G = ( iceControlling && local . priority ) || remote . priority ; const D = ( iceControlling && remote . priority ) || local . priority ; return ( 1 << 32 ) * Math . min ( G , D ) + 2 * Math . max ( G , D ) + ( G > D ? 1 : 0 ) ; } export function getHostAddress ( useIpv4 , useIpv6 ) { const address = [ ] ; if ( useIpv4 ) address . push ( nodeIp . address ( "" , "ipv4" ) ) ; if ( useIpv6 ) address . push ( nodeIp . address ( "" , "ipv6" ) ) ; return address ; } export async function serverReflexiveCandidate ( protocol , stunServer ) { const request = new Message ( methods . BINDING , classes . REQUEST ) ; try { const [ response ] = await protocol . request ( request , stunServer ) ; const localCandidate = protocol . localCandidate ; if ( ! localCandidate ) throw new Error ( ) ; return new Candidate ( candidateFoundation ( "srflx" , "udp" , localCandidate . host ) , localCandidate . component , localCandidate . transport , candidatePriority ( localCandidate . component , "srflx" ) , response . attributes [ "XOR-MAPPED-ADDRESS" ] [ 0 ] , response . attributes [ "XOR-MAPPED-ADDRESS" ] [ 1 ] , "srflx" , localCandidate . host , localCandidate . port ) ; } catch ( error ) { } }