import { EventEmitter } from 'events' ; import { rlp } from 'ethereumjs-util' ; import ms from 'ms' ; import snappy from 'snappyjs' ; import { debug as createDebugLogger } from 'debug' ; import { int2buffer , buffer2int , assertEq , formatLogData } from '../util' ; import { Peer , DISCONNECT_REASONS } from '../rlpx/peer' ; const DEBUG_BASE_NAME = 'devp2p:les' ; const debug = createDebugLogger ( DEBUG_BASE_NAME ) ; const verbose = createDebugLogger ( 'verbose' ) . enabled ; export const DEFAULT_ANNOUNCE_TYPE = 1 ; /**
 * Will be set to the first successfully connected peer to allow for
 * debugging with the `devp2p:FIRST_PEER` debugger
 */ let _firstPeer = '' ; type SendMethod = ( code : LES . MESSAGE_CODES , data : Buffer ) => any ; export class LES extends EventEmitter { _version : number ; _peer : Peer ; _send : SendMethod ; _status : LES . Status | null ; _peerStatus : LES . Status | null ; _statusTimeoutId : NodeJS . Timeout ; private msgDebuggers : { [ key ] : ( debug : string ) => void ; } = { } ; constructor ( version , peer , send ) { super ( ) ; this . _version = version ; this . _peer = peer ; this . _send = send ; this . _status = null ; this . _peerStatus = null ; this . _statusTimeoutId = setTimeout ( ( ) => { this . _peer . disconnect ( DISCONNECT_REASONS . TIMEOUT ) ; } , ms ( '5s' ) ) ; this . initMsgDebuggers ( ) ; } static les2 = { name : 'les' , version : 2 , length : 21 , constructor : LES } ; static les3 = { name : 'les' , version : 3 , length : 23 , constructor : LES } ; static les4 = { name : 'les' , version : 4 , length : 23 , constructor : LES } ; _handleMessage ( code , data ) { const payload = rlp . decode ( data ) ; const messageName = this . getMsgPrefix ( code ) ; const debugMsg = `Received ${ messageName } message from ${ this . _peer . _socket . remoteAddress }:${ this . _peer . _socket . remotePort }` ; if ( code !== LES . MESSAGE_CODES . STATUS ) { const logData = formatLogData ( data . toString ( 'hex' ) , verbose ) ; this . debug ( messageName , `${ debugMsg }: ${ logData }` ) ; } switch ( code ) { case LES . MESSAGE_CODES . STATUS : { assertEq ( this . _peerStatus , null , 'Uncontrolled status message' , this . debug . bind ( this ) , 'STATUS' ) ; const statusArray = { } ; payload . forEach ( function ( value ) { statusArray [ value [ 0 ] . toString ( ) ] = value [ 1 ] ; } ) ; this . _peerStatus = statusArray ; const peerStatusMsg = `${ this . _peerStatus ? this . _getStatusString ( this . _peerStatus ) : '' }` ; this . debug ( messageName , `${ debugMsg }: ${ peerStatusMsg }` ) ; this . _handleStatus ( ) ; break ; } case LES . MESSAGE_CODES . ANNOUNCE : case LES . MESSAGE_CODES . GET_BLOCK_HEADERS : case LES . MESSAGE_CODES . BLOCK_HEADERS : case LES . MESSAGE_CODES . GET_BLOCK_BODIES : case LES . MESSAGE_CODES . BLOCK_BODIES : case LES . MESSAGE_CODES . GET_RECEIPTS : case LES . MESSAGE_CODES . RECEIPTS : case LES . MESSAGE_CODES . GET_PROOFS : case LES . MESSAGE_CODES . PROOFS : case LES . MESSAGE_CODES . GET_CONTRACT_CODES : case LES . MESSAGE_CODES . CONTRACT_CODES : case LES . MESSAGE_CODES . GET_HEADER_PROOFS : case LES . MESSAGE_CODES . HEADER_PROOFS : case LES . MESSAGE_CODES . SEND_TX : case LES . MESSAGE_CODES . GET_PROOFS_V2 : case LES . MESSAGE_CODES . PROOFS_V2 : case LES . MESSAGE_CODES . GET_HELPER_TRIE_PROOFS : case LES . MESSAGE_CODES . HELPER_TRIE_PROOFS : case LES . MESSAGE_CODES . SEND_TX_V2 : case LES . MESSAGE_CODES . GET_TX_STATUS : case LES . MESSAGE_CODES . TX_STATUS : if ( this . _version >= LES . les2 . version ) break ; return ; case LES . MESSAGE_CODES . STOP_MSG : case LES . MESSAGE_CODES . RESUME_MSG : if ( this . _version >= LES . les3 . version ) break ; return ; default : return ; } this . emit ( 'message' , code , payload ) ; } _handleStatus ( ) { if ( this . _status === null || this . _peerStatus === null ) return ; clearTimeout ( this . _statusTimeoutId ) ; assertEq ( this . _status [ 'protocolVersion' ] , this . _peerStatus [ 'protocolVersion' ] , 'Protocol version mismatch' , this . debug . bind ( this ) , 'STATUS' ) ; assertEq ( this . _status [ 'networkId' ] , this . _peerStatus [ 'networkId' ] , 'NetworkId mismatch' , this . debug . bind ( this ) , 'STATUS' ) ; assertEq ( this . _status [ 'genesisHash' ] , this . _peerStatus [ 'genesisHash' ] , 'Genesis block mismatch' , this . debug . bind ( this ) , 'STATUS' ) ; this . emit ( 'status' , this . _peerStatus ) ; if ( _firstPeer === '' ) { this . _addFirstPeerDebugger ( ) ; } } getVersion ( ) { return this . _version ; } _getStatusString ( status ) { let sStr = `[V:${ buffer2int ( status [ 'protocolVersion' ] ) }, ` ; sStr += `NID:${ buffer2int ( status [ 'networkId' ] as Buffer ) }, HTD:${ buffer2int ( status [ 'headTd' ] ) }, ` ; sStr += `HeadH:${ status [ 'headHash' ] . toString ( 'hex' ) }, HeadN:${ buffer2int ( status [ 'headNum' ] ) }, ` ; sStr += `GenH:${ status [ 'genesisHash' ] . toString ( 'hex' ) }` ; if ( status [ 'serveHeaders' ] ) sStr += `, serveHeaders active` ; if ( status [ 'serveChainSince' ] ) sStr += `, ServeCS: ${ buffer2int ( status [ 'serveChainSince' ] ) }` ; if ( status [ 'serveStateSince' ] ) sStr += `, ServeSS: ${ buffer2int ( status [ 'serveStateSince' ] ) }` ; if ( status [ 'txRelay' ] ) sStr += `, txRelay active` ; if ( status [ 'flowControl/BL' ] ) sStr += `, flowControl/BL set` ; if ( status [ 'flowControl/MRR' ] ) sStr += `, flowControl/MRR set` ; if ( status [ 'flowControl/MRC' ] ) sStr += `, flowControl/MRC set` ; if ( status [ 'forkID' ] ) sStr += `, forkID: [crc32: ${ status [ 'forkID' ] [ 0 ] . toString ( 'hex' ) }, nextFork: ${ buffer2int ( status [ 'forkID' ] [ 1 ] ) }]` ; if ( status [ 'recentTxLookup' ] ) sStr += `, recentTxLookup: ${ buffer2int ( status [ 'recentTxLookup' ] ) }` ; sStr += `]` ; return sStr ; } sendStatus ( status ) { if ( this . _status !== null ) return ; if ( ! status . announceType ) { status [ 'announceType' ] = int2buffer ( DEFAULT_ANNOUNCE_TYPE ) ; } status [ 'protocolVersion' ] = int2buffer ( this . _version ) ; status [ 'networkId' ] = this . _peer . _common . chainIdBN ( ) . toArrayLike ( Buffer ) ; this . _status = status ; const statusList = [ ] ; Object . keys ( status ) . forEach ( ( key ) => { statusList . push ( [ key , status [ key ] ] ) ; } ) ; this . debug ( 'STATUS' , `Send STATUS message to ${ this . _peer . _socket . remoteAddress }:${ this . _peer . _socket . remotePort } (les${ this . _version }): ${ this . _getStatusString ( this . _status ) }` ) ; let payload = rlp . encode ( statusList ) ; if ( this . _peer . _hello ?. protocolVersion && this . _peer . _hello ?. protocolVersion >= 5 ) { payload = snappy . compress ( payload ) ; } this . _send ( LES . MESSAGE_CODES . STATUS , payload ) ; this . _handleStatus ( ) ; } param code param payload sendMessage ( code , payload ) { const messageName = this . getMsgPrefix ( code ) ; const logData = formatLogData ( rlp . encode ( payload ) . toString ( 'hex' ) , verbose ) ; const debugMsg = `Send ${ messageName } message to ${ this . _peer . _socket . remoteAddress }:${ this . _peer . _socket . remotePort }: ${ logData }` ; this . debug ( messageName , debugMsg ) ; switch ( code ) { case LES . MESSAGE_CODES . STATUS : throw new Error ( 'Please send status message through .sendStatus' ) ; case LES . MESSAGE_CODES . ANNOUNCE : case LES . MESSAGE_CODES . GET_BLOCK_HEADERS : case LES . MESSAGE_CODES . BLOCK_HEADERS : case LES . MESSAGE_CODES . GET_BLOCK_BODIES : case LES . MESSAGE_CODES . BLOCK_BODIES : case LES . MESSAGE_CODES . GET_RECEIPTS : case LES . MESSAGE_CODES . RECEIPTS : case LES . MESSAGE_CODES . GET_PROOFS : case LES . MESSAGE_CODES . PROOFS : case LES . MESSAGE_CODES . GET_CONTRACT_CODES : case LES . MESSAGE_CODES . CONTRACT_CODES : case LES . MESSAGE_CODES . GET_HEADER_PROOFS : case LES . MESSAGE_CODES . HEADER_PROOFS : case LES . MESSAGE_CODES . SEND_TX : case LES . MESSAGE_CODES . GET_PROOFS_V2 : case LES . MESSAGE_CODES . PROOFS_V2 : case LES . MESSAGE_CODES . GET_HELPER_TRIE_PROOFS : case LES . MESSAGE_CODES . HELPER_TRIE_PROOFS : case LES . MESSAGE_CODES . SEND_TX_V2 : case LES . MESSAGE_CODES . GET_TX_STATUS : case LES . MESSAGE_CODES . TX_STATUS : if ( this . _version >= LES . les2 . version ) break ; throw new Error ( `Code ${ code } not allowed with version ${ this . _version }` ) ; case LES . MESSAGE_CODES . STOP_MSG : case LES . MESSAGE_CODES . RESUME_MSG : if ( this . _version >= LES . les3 . version ) break ; throw new Error ( `Code ${ code } not allowed with version ${ this . _version }` ) ; default : throw new Error ( `Unknown code ${ code }` ) ; } payload = rlp . encode ( payload ) ; if ( this . _peer . _hello ?. protocolVersion && this . _peer . _hello ?. protocolVersion >= 5 ) { payload = snappy . compress ( payload ) ; } this . _send ( code , payload ) ; } getMsgPrefix ( msgCode ) { return LES . MESSAGE_CODES [ msgCode ] ; } private initMsgDebuggers ( ) { const MESSAGE_NAMES = Object . values ( LES . MESSAGE_CODES ) . filter ( ( value ) => typeof value === 'string' ) as string [ ] ; for ( const name of MESSAGE_NAMES ) { this . msgDebuggers [ name ] = createDebugLogger ( `${ DEBUG_BASE_NAME }:${ name }` ) ; } } /**
     * Called once on the peer where a first successful `STATUS`
     * msg exchange could be achieved.
     *
     * Can be used together with the `devp2p:FIRST_PEER` debugger.
     */ _addFirstPeerDebugger ( ) { const ip = this . _peer . _socket . remoteAddress ; if ( ip ) { this . msgDebuggers [ ip ] = createDebugLogger ( `devp2p:FIRST_PEER` ) ; this . _peer . _addFirstPeerDebugger ( ) ; _firstPeer = ip ; } } param messageName param msg private debug ( messageName , msg ) { debug ( msg ) ; if ( this . msgDebuggers [ messageName ] ) { this . msgDebuggers [ messageName ] ( msg ) ; } } } export namespace LES { export interface Status { [ key ] : any ; protocolVersion : Buffer ; networkId : Buffer ; headTd : Buffer ; headHash : Buffer ; headNum : Buffer ; genesisHash : Buffer ; serveHeaders : Buffer ; serveChainSince : Buffer ; serveStateSince : Buffer ; txRelay : Buffer ; 'flowControl/BL' : Buffer ; 'flowControl/MRR' : Buffer ; 'flowControl/MRC' : Buffer ; announceType : Buffer ; forkID : [ Buffer , Buffer ] ; recentTxLookup : Buffer ; } export enum MESSAGE_CODES { STATUS = 0x00 , ANNOUNCE = 0x01 , GET_BLOCK_HEADERS = 0x02 , BLOCK_HEADERS = 0x03 , GET_BLOCK_BODIES = 0x04 , BLOCK_BODIES = 0x05 , GET_RECEIPTS = 0x06 , RECEIPTS = 0x07 , GET_PROOFS = 0x08 , PROOFS = 0x09 , GET_CONTRACT_CODES = 0x0a , CONTRACT_CODES = 0x0b , GET_HEADER_PROOFS = 0x0d , HEADER_PROOFS = 0x0e , SEND_TX = 0x0c , GET_PROOFS_V2 = 0x0f , PROOFS_V2 = 0x10 , GET_HELPER_TRIE_PROOFS = 0x11 , HELPER_TRIE_PROOFS = 0x12 , SEND_TX_V2 = 0x13 , GET_TX_STATUS = 0x14 , TX_STATUS = 0x15 , STOP_MSG = 0x16 , RESUME_MSG = 0x17 } }