import { EventEmitter } from 'events' ; import * as IlpPacket from 'ilp-packet' ; import * as ILDCP from 'ilp-protocol-ildcp' ; import createLogger from 'ilp-logger' ; import * as cryptoHelper from './crypto' ; import { Connection , ConnectionOpts } from './connection' ; import { ServerConnectionPool } from './pool' ; import { Predictor , Writer } from 'oer-utils' ; import { Plugin } from './util/plugin-interface' ; const DEFAULT_DISCONNECT_DELAY = 100 ; export interface ServerOpts extends ConnectionOpts { serverSecret ? : Buffer ; /**
     * Number of milliseconds to wait between closing the server and disconnecting
     * the plugin so packets may be safely returned
     */ disconnectDelay ? : number ; } export interface GenerateAddressSecretOpts { connectionTag ? : string ; receiptNonce ? : Buffer ; receiptSecret ? : Buffer ; } /**
 * STREAM Server that can listen on an account and handle multiple incoming [`Connection`s] Connection .
 * Note: the connections this refers to are over ILP, not over the Internet.
 *
 * The Server operator should give a unique address and secret (generated by calling
 * [`generateAddressAndSecret`] generateAddressAndSecret ) to each client that it expects to connect.
 *
 * The Server will emit a `'connection'` event when the first packet is received for a specific Connection.
  export class Server extends EventEmitter { protected serverSecret : Buffer ; protected plugin : Plugin ; protected serverAccount : string ; protected serverAssetCode : string ; protected serverAssetScale : number ; protected log : any ; protected enablePadding ? : boolean ; protected connected : boolean ; protected connectionOpts : ConnectionOpts ; protected pendingRequests : Promise < any > = Promise . resolve ( ) ; protected disconnectDelay : number ; private pool : ServerConnectionPool ; constructor ( opts ) { super ( ) ; this . serverSecret = opts . serverSecret || cryptoHelper . randomBytes ( 32 ) ; this . plugin = opts . plugin ; this . log = createLogger ( 'ilp-protocol-stream:Server' ) ; this . connectionOpts = Object . assign ( { } , opts , { serverSecret : undefined } ) as ConnectionOpts ; this . disconnectDelay = opts . disconnectDelay || DEFAULT_DISCONNECT_DELAY ; this . connected = false ; } /**
     * Event fired when a new [`Connection`] Connection is received.
     * The connection event handler should immediately (synchronously) add a
     * `"stream"` event handler to ensure no incoming streams are ignored.
     *  event type { Connection } fires async listen ( ) { if ( this . connected && this . plugin . isConnected ( ) ) { return ; } this . plugin . registerDataHandler ( data => { this . emit ( '_incoming_prepare' ) ; const request = this . handleData ( data ) ; this . pendingRequests = this . pendingRequests . then ( ( ) => request . finally ( ) ) ; return request ; } ) ; await this . plugin . connect ( ) ; const { clientAddress , assetCode , assetScale } = await ILDCP . fetch ( this . plugin . sendData . bind ( this . plugin ) ) ; this . serverAccount = clientAddress ; this . serverAssetCode = assetCode ; this . serverAssetScale = assetScale ; this . connected = true ; this . pool = new ServerConnectionPool ( this . serverSecret , { ... this . connectionOpts , isServer : true , plugin : this . plugin , sourceAccount : this . serverAccount , assetCode : this . serverAssetCode , assetScale : this . serverAssetScale } , ( connection ) => { this . emit ( 'connection' , connection ) ; } ) ; } /**
     * End all connections and disconnect the plugin
     */ async close ( ) { this . plugin . deregisterDataHandler ( ) ; this . plugin . registerDataHandler ( async ( ) => IlpPacket . serializeIlpReject ( { code : IlpPacket . Errors . codes . T99_APPLICATION_ERROR , triggeredBy : this . serverAccount , message : 'Shutting down server' , data : Buffer . alloc ( 0 ) } ) ) ; await this . pendingRequests ; await new Promise ( r => setTimeout ( r , this . disconnectDelay ) ) ; await this . pool . close ( ) ; this . plugin . deregisterDataHandler ( ) ; await this . plugin . disconnect ( ) ; this . emit ( '_close' ) ; this . connected = false ; } /**
     * Resolves when the next connection is accepted.
     *
     * To handle subsequent connections, the user must call `acceptConnection` again.
     * Alternatively, the user can listen on the `'connection'` event.
     */ async acceptConnection ( ) { await this . listen ( ) ; return new Promise ( ( resolve , reject ) => { const done = ( connection ) => { this . removeListener ( 'connection' , done ) ; this . removeListener ( '_close' , done ) ; if ( connection ) resolve ( connection ) ; else reject ( new Error ( 'server closed' ) ) ; } ; this . once ( 'connection' , done ) ; this . once ( '_close' , done ) ; } ) as Promise < Connection > ; } param connectionTag param receiptNonce param receiptSecret generateAddressAndSecret ( opts ? ) { if ( ! this . connected ) { throw new Error ( 'Server must be connected to generate address and secret' ) ; } let connectionTag = Buffer . alloc ( 0 ) ; let receiptNonce = Buffer . alloc ( 0 ) ; let receiptSecret = Buffer . alloc ( 0 ) ; let receiptsEnabled = false ; if ( opts ) { if ( typeof opts === 'object' ) { if ( opts . connectionTag ) { connectionTag = Buffer . from ( opts . connectionTag , 'ascii' ) ; } if ( ! opts . receiptNonce !== ! opts . receiptSecret ) { throw new Error ( 'receiptNonce and receiptSecret must accompany each other' ) ; } if ( opts . receiptNonce ) { if ( opts . receiptNonce . length !== 16 ) { throw new Error ( 'receiptNonce must be 16 bytes' ) ; } receiptsEnabled = true ; receiptNonce = opts . receiptNonce ; } if ( opts . receiptSecret ) { if ( opts . receiptSecret . length !== 32 ) { throw new Error ( 'receiptSecret must be 32 bytes' ) ; } receiptSecret = opts . receiptSecret ; } } else { connectionTag = Buffer . from ( opts , 'ascii' ) ; } } const tokenNonce = cryptoHelper . generateTokenNonce ( ) ; const predictor = new Predictor ( ) ; predictor . writeOctetString ( tokenNonce , cryptoHelper . TOKEN_NONCE_LENGTH ) ; predictor . writeVarOctetString ( connectionTag ) ; predictor . writeVarOctetString ( receiptNonce ) ; predictor . writeVarOctetString ( receiptSecret ) ; const writer = new Writer ( predictor . length ) ; writer . writeOctetString ( tokenNonce , cryptoHelper . TOKEN_NONCE_LENGTH ) ; writer . writeVarOctetString ( connectionTag ) ; writer . writeVarOctetString ( receiptNonce ) ; writer . writeVarOctetString ( receiptSecret ) ; const token = cryptoHelper . encryptConnectionAddressToken ( this . serverSecret , writer . getBuffer ( ) ) ; const sharedSecret = cryptoHelper . generateSharedSecretFromToken ( this . serverSecret , token ) ; return { destinationAccount : `${ this . serverAccount }.${ base64url ( token ) }` , sharedSecret , receiptsEnabled } ; } get assetCode ( ) : string { if ( ! this . connected ) { throw new Error ( 'Server must be connected to get asset code.' ) ; } return this . serverAssetCode ; } get assetScale ( ) : number { if ( ! this . connected ) { throw new Error ( 'Server must be connected to get asset scale.' ) ; } return this . serverAssetScale ; } /**
     * Parse incoming ILP Prepare packets and pass them to the correct connection
     */ protected async handleData ( data ) { try { let prepare ; try { prepare = IlpPacket . deserializeIlpPrepare ( data ) ; } catch ( err ) { this . log . error ( 'got data that is not an ILP Prepare packet: %h' , data ) ; return IlpPacket . serializeIlpReject ( { code : 'F00' , message : `Expected an ILP Prepare packet (type 12), but got packet with type: ${ data [ 0 ] }` , data : Buffer . alloc ( 0 ) , triggeredBy : this . serverAccount } ) ; } const localAddressParts = prepare . destination . replace ( this . serverAccount + '.' , '' ) . split ( '.' ) ; if ( localAddressParts . length === 0 || ! localAddressParts [ 0 ] ) { this . log . error ( 'destination in ILP Prepare packet does not have a Connection ID: %s' , prepare . destination ) ; throw new IlpPacket . Errors . UnreachableError ( '' ) ; } const connectionId = localAddressParts [ 0 ] ; const connection = await this . pool . getConnection ( connectionId , prepare ) . catch ( ( _err ) => { throw new IlpPacket . Errors . UnreachableError ( '' ) ; } ) ; const fulfill = await connection . handlePrepare ( prepare ) ; return IlpPacket . serializeIlpFulfill ( fulfill ) ; } catch ( err ) { if ( ! err . ilpErrorCode ) { this . log . error ( 'error handling prepare:' , err ) ; } return IlpPacket . serializeIlpReject ( { code : err . ilpErrorCode || 'F00' , message : err . ilpErrorMessage || '' , data : err . ilpErrorData || Buffer . alloc ( 0 ) , triggeredBy : this . serverAccount || '' } ) ; } } } /**
 * Creates a [`Server`] Server and resolves when the server is connected and listening
  export async function createServer ( opts ) { const server = new Server ( opts ) ; await server . listen ( ) ; return server ; } function base64url ( buffer ) { return buffer . toString ( 'base64' ) . replace ( /=+$/ , '' ) . replace ( /\+/g , '-' ) . replace ( /\//g , '_' ) ; }