import assert from 'assert' ; import { EventEmitter } from 'events' ; import * as rlp from 'rlp' ; import ms from 'ms' ; import { int2buffer , buffer2int , assertEq , formatLogId , formatLogData } from '../util' ; import { Peer , DISCONNECT_REASONS } from '../rlpx/peer' ; import { debug as createDebugLogger } from 'debug' ; const debug = createDebugLogger ( 'devp2p:eth' ) ; const verbose = createDebugLogger ( 'verbose' ) . enabled ; type SendMethod = ( code : ETH . MESSAGE_CODES , data : Buffer ) => any ; export class ETH extends EventEmitter { _version : number ; _peer : Peer ; _status : ETH . StatusMsg | null ; _peerStatus : ETH . StatusMsg | null ; _statusTimeoutId : NodeJS . Timeout ; _send : SendMethod ; _hardfork : string = 'chainstart' ; _latestBlock : number = 0 ; _forkHash : string = '' ; _nextForkBlock : number = 0 ; 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' ) ) ; if ( this . _version >= 64 ) { const c = this . _peer . _common ; this . _hardfork = c . hardfork ( ) ? ( c . hardfork ( ) as string ) : this . _hardfork ; this . _latestBlock = c . hardforkBlock ( this . _hardfork ) ; this . _forkHash = c . forkHash ( this . _hardfork ) ; const nextForkBlock = c . nextHardforkBlock ( this . _hardfork ) ; this . _nextForkBlock = nextForkBlock ? nextForkBlock : 0 ; } } static eth62 = { name : 'eth' , version : 62 , length : 8 , constructor : ETH } ; static eth63 = { name : 'eth' , version : 63 , length : 17 , constructor : ETH } ; static eth64 = { name : 'eth' , version : 64 , length : 29 , constructor : ETH } ; _handleMessage ( code , data ) { const payload = rlp . decode ( data ) as unknown ; if ( code !== ETH . MESSAGE_CODES . STATUS ) { const debugMsg = `Received ${ this . getMsgPrefix ( code ) } message from ${ this . _peer . _socket . remoteAddress }:${ this . _peer . _socket . remotePort }` ; const logData = formatLogData ( data . toString ( 'hex' ) , verbose ) ; debug ( `${ debugMsg }: ${ logData }` ) ; } switch ( code ) { case ETH . MESSAGE_CODES . STATUS : assertEq ( this . _peerStatus , null , 'Uncontrolled status message' , debug ) ; this . _peerStatus = payload as ETH . StatusMsg ; debug ( `Received ${ this . getMsgPrefix ( code ) } message from ${ this . _peer . _socket . remoteAddress }:${ this . _peer . _socket . remotePort }: : ${ this . _peerStatus ? this . _getStatusString ( this . _peerStatus ) : '' }` ) ; this . _handleStatus ( ) ; break ; case ETH . MESSAGE_CODES . NEW_BLOCK_HASHES : case ETH . MESSAGE_CODES . TX : case ETH . MESSAGE_CODES . GET_BLOCK_HEADERS : case ETH . MESSAGE_CODES . BLOCK_HEADERS : case ETH . MESSAGE_CODES . GET_BLOCK_BODIES : case ETH . MESSAGE_CODES . BLOCK_BODIES : case ETH . MESSAGE_CODES . NEW_BLOCK : if ( this . _version >= ETH . eth62 . version ) break ; return ; case ETH . MESSAGE_CODES . GET_NODE_DATA : case ETH . MESSAGE_CODES . NODE_DATA : case ETH . MESSAGE_CODES . GET_RECEIPTS : case ETH . MESSAGE_CODES . RECEIPTS : if ( this . _version >= ETH . eth63 . version ) break ; return ; default : return ; } this . emit ( 'message' , code , payload ) ; } param forkId _validateForkId ( forkId ) { const c = this . _peer . _common ; const peerForkHash = `0x${ forkId [ 0 ] . toString ( 'hex' ) }` ; const peerNextFork = buffer2int ( forkId [ 1 ] ) ; if ( this . _forkHash === peerForkHash ) { if ( peerNextFork !== 0 ) { if ( this . _latestBlock >= peerNextFork ) { const msg = 'Remote is advertising a future fork that passed locally' ; debug ( msg ) ; throw new assert . AssertionError ( { message : msg } ) ; } } } const peerFork = c . hardforkForForkHash ( peerForkHash ) ; if ( peerFork === null ) { const msg = 'Unknown fork hash' ; debug ( msg ) ; throw new assert . AssertionError ( { message : msg } ) ; } if ( ! c . hardforkGteHardfork ( peerFork . name , this . _hardfork ) ) { if ( peerNextFork === null || c . nextHardforkBlock ( peerFork . name ) !== peerNextFork ) { const msg = 'Outdated fork status, remote needs software update' ; debug ( msg ) ; throw new assert . AssertionError ( { message : msg } ) ; } } } _handleStatus ( ) { if ( this . _status === null || this . _peerStatus === null ) return ; clearTimeout ( this . _statusTimeoutId ) ; assertEq ( this . _status [ 0 ] , this . _peerStatus [ 0 ] , 'Protocol version mismatch' , debug ) ; assertEq ( this . _status [ 1 ] , this . _peerStatus [ 1 ] , 'NetworkId mismatch' , debug ) ; assertEq ( this . _status [ 4 ] , this . _peerStatus [ 4 ] , 'Genesis block mismatch' , debug ) ; const status = { networkId : this . _peerStatus [ 1 ] , td : Buffer . from ( this . _peerStatus [ 2 ] ) , bestHash : Buffer . from ( this . _peerStatus [ 3 ] ) , genesisHash : Buffer . from ( this . _peerStatus [ 4 ] ) } ; if ( this . _version >= 64 ) { assertEq ( this . _peerStatus [ 5 ] . length , 2 , 'Incorrect forkId msg format' , debug ) ; this . _validateForkId ( this . _peerStatus [ 5 ] as Buffer [ ] ) ; status [ 'forkId' ] = this . _peerStatus [ 5 ] ; } this . emit ( 'status' , status ) ; } getVersion ( ) { return this . _version ; } _forkHashFromForkId ( forkId ) { return `0x${ forkId . toString ( 'hex' ) }` ; } _nextForkFromForkId ( forkId ) { return buffer2int ( forkId ) ; } _getStatusString ( status ) { let sStr = `[V:${ buffer2int ( status [ 0 ] as Buffer ) }, NID:${ buffer2int ( status [ 1 ] as Buffer ) }, TD:${ buffer2int ( status [ 2 ] as Buffer ) }` ; sStr += `, BestH:${ formatLogId ( status [ 3 ] . toString ( 'hex' ) , verbose ) }, GenH:${ formatLogId ( status [ 4 ] . toString ( 'hex' ) , verbose ) }` ; if ( this . _version >= 64 ) { sStr += `, ForkHash: 0x${ ( status [ 5 ] [ 0 ] as Buffer ) . toString ( 'hex' ) }` ; sStr += `, ForkNext: ${ buffer2int ( status [ 5 ] [ 1 ] as Buffer ) }` ; } sStr += `]` ; return sStr ; } sendStatus ( status ) { if ( this . _status !== null ) return ; this . _status = [ int2buffer ( this . _version ) , int2buffer ( this . _peer . _common . chainId ( ) ) , status . td , status . bestHash , status . genesisHash ] ; if ( this . _version >= 64 ) { if ( status . latestBlock ) { if ( status . latestBlock < this . _latestBlock ) { throw new Error ( 'latest block provided is not matching the HF setting of the Common instance (Rlpx)' ) ; } this . _latestBlock = status . latestBlock ; } const forkHashB = Buffer . from ( this . _forkHash . substr ( 2 ) , 'hex' ) ; const nextForkB = Buffer . from ( this . _nextForkBlock . toString ( 16 ) , 'hex' ) ; this . _status . push ( [ forkHashB , nextForkB ] ) ; } debug ( `Send STATUS message to ${ this . _peer . _socket . remoteAddress }:${ this . _peer . _socket . remotePort } (eth${ this . _version }): ${ this . _getStatusString ( this . _status ) }` ) ; this . _send ( ETH . MESSAGE_CODES . STATUS , rlp . encode ( this . _status as any ) ) ; this . _handleStatus ( ) ; } sendMessage ( code , payload ) { const debugMsg = `Send ${ this . getMsgPrefix ( code ) } message to ${ this . _peer . _socket . remoteAddress }:${ this . _peer . _socket . remotePort }` ; const logData = formatLogData ( rlp . encode ( payload ) . toString ( 'hex' ) , verbose ) ; debug ( `${ debugMsg }: ${ logData }` ) ; switch ( code ) { case ETH . MESSAGE_CODES . STATUS : throw new Error ( 'Please send status message through .sendStatus' ) ; case ETH . MESSAGE_CODES . NEW_BLOCK_HASHES : case ETH . MESSAGE_CODES . TX : case ETH . MESSAGE_CODES . GET_BLOCK_HEADERS : case ETH . MESSAGE_CODES . BLOCK_HEADERS : case ETH . MESSAGE_CODES . GET_BLOCK_BODIES : case ETH . MESSAGE_CODES . BLOCK_BODIES : case ETH . MESSAGE_CODES . NEW_BLOCK : if ( this . _version >= ETH . eth62 . version ) break ; throw new Error ( `Code ${ code } not allowed with version ${ this . _version }` ) ; case ETH . MESSAGE_CODES . GET_NODE_DATA : case ETH . MESSAGE_CODES . NODE_DATA : case ETH . MESSAGE_CODES . GET_RECEIPTS : case ETH . MESSAGE_CODES . RECEIPTS : if ( this . _version >= ETH . eth63 . version ) break ; throw new Error ( `Code ${ code } not allowed with version ${ this . _version }` ) ; default : throw new Error ( `Unknown code ${ code }` ) ; } this . _send ( code , rlp . encode ( payload ) ) ; } getMsgPrefix ( msgCode ) { return ETH . MESSAGE_CODES [ msgCode ] ; } } export namespace ETH { export interface StatusMsg extends Array < Buffer | Buffer [ ] > { } export type StatusOpts = { version : number ; td : Buffer ; bestHash : Buffer ; latestBlock ? : number ; genesisHash : Buffer ; } ; export enum MESSAGE_CODES { STATUS = 0x00 , NEW_BLOCK_HASHES = 0x01 , TX = 0x02 , GET_BLOCK_HEADERS = 0x03 , BLOCK_HEADERS = 0x04 , GET_BLOCK_BODIES = 0x05 , BLOCK_BODIES = 0x06 , NEW_BLOCK = 0x07 , GET_NODE_DATA = 0x0d , NODE_DATA = 0x0e , GET_RECEIPTS = 0x0f , RECEIPTS = 0x10 } }