format import { SecureServerConfig } from '../utils/CertificateProvider' ; import ServerAdapter , { SecureClientQuery , ServerEventsListener , } from './ServerAdapter' ; import tls from 'tls' ; import net , { AddressInfo , Socket } from 'net' ; import { RSocketServer } from 'rsocket-core' ; import RSocketTCPServer from 'rsocket-tcp-server' ; import { Payload , ReactiveSocket , Responder } from 'rsocket-types' ; import { Single } from 'rsocket-flowable' ; import { ClientConnection , ConnectionStatusChange , ConnectionStatus , } from './ClientConnection' ; import { ClientDescription , ClientQuery , ClientResponseType , } from 'flipper-common' ; /**
 * RSocket based server. RSocket uses its own protocol for communication between
 * client and server.
 */ class ServerRSocket extends ServerAdapter { rawServer_ : RSocketServer < any , any > | null | undefined ; constructor ( listener ) { super ( listener ) ; this . rawServer_ = null ; } /**
     * Start the server bound to the specified port. It configures
     * the RSocket server factory and request handler based on the optional
     * sslConfig argument.
     */ start ( port , sslConfig ? ) { const self = this ; return new Promise ( ( resolve , reject ) => { let rawServer ; const serverFactory = ( onConnect ) => { const transportServer = sslConfig ? tls . createServer ( sslConfig , ( socket ) => { onConnect ( socket ) ; } ) : net . createServer ( onConnect ) ; transportServer . on ( 'error' , reject ) . on ( 'listening' , ( ) => { console . debug ( `${ sslConfig ? 'Secure' : 'Certificate' } server started on port ${ port }` , 'server' ) ; self . listener . onListening ( port ) ; self . rawServer_ = rawServer ; resolve ( ( transportServer . address ( ) as AddressInfo ) . port ) ; } ) ; return transportServer ; } ; rawServer = new RSocketServer ( { getRequestHandler : sslConfig ? this . _trustedRequestHandler : this . _untrustedRequestHandler , transport : new RSocketTCPServer ( { port : port , serverFactory : serverFactory , } ) , } ) ; rawServer . start ( ) ; } ) ; } stop ( ) { if ( this . rawServer_ ) { return Promise . resolve ( this . rawServer_ . stop ( ) ) ; } return Promise . resolve ( ) ; } param socket param payload returns _trustedRequestHandler = ( socket , payload ) : Partial < Responder < string , any > > => { if ( ! payload . data ) { return { } ; } const clientQuery = JSON . parse ( payload . data ) ; this . listener . onSecureConnectionAttempt ( clientQuery ) ; console . info ( `[conn] Secure rsocket connection attempt: ${ clientQuery . app } on ${ clientQuery . device_id }. Medium ${ clientQuery . medium }. CSR: ${ clientQuery . csr_path }` ) ; const clientConnection = { subscribeToEvents ( subscriber ) { socket . connectionStatus ( ) . subscribe ( { onNext ( payload ) { let status = ConnectionStatus . CONNECTED ; if ( payload . kind == 'ERROR' ) status = ConnectionStatus . ERROR ; else if ( payload . kind == 'CLOSED' ) status = ConnectionStatus . CLOSED ; else if ( payload . kind == 'CONNECTED' ) status = ConnectionStatus . CONNECTED ; else if ( payload . kind == 'NOT_CONNECTED' ) status = ConnectionStatus . NOT_CONNECTED ; else if ( payload . kind == 'CONNECTING' ) status = ConnectionStatus . CONNECTING ; subscriber ( status ) ; } , onSubscribe ( subscription ) { subscription . request ( Number . MAX_SAFE_INTEGER ) ; } , onError ( payload ) { console . error ( '[client] connection status error ' , payload ) ; } , } ) ; } , close ( ) { socket . close ( ) ; } , send ( data ) { socket . fireAndForget ( { data : JSON . stringify ( data ) } ) ; } , sendExpectResponse ( data ) { return new Promise < any > ( ( resolve , reject ) => { socket . requestResponse ( { data : JSON . stringify ( data ) , } ) . subscribe ( { onComplete : ( payload ) => { const response = JSON . parse ( payload . data ) ; response . length = payload . data . length ; resolve ( response ) ; } , onError : ( e ) => { reject ( e ) ; } , } ) ; } ) ; } , } ; let resolvedClient ; const client = this . listener . onConnectionCreated ( clientQuery , clientConnection ) ; client . then ( ( client ) => { console . info ( `[conn] Client connected: ${ clientQuery . app } on ${ clientQuery . device_id }. Medium ${ clientQuery . medium }. CSR: ${ clientQuery . csr_path }` ) ; resolvedClient = client ; } ) . catch ( ( e ) => { console . error ( '[conn] Failed to resolve new client' , e ) ; } ) ; return { fireAndForget : ( payload ) => { if ( resolvedClient ) { this . listener . onClientMessage ( resolvedClient . id , payload . data ) ; } else { client && client . then ( ( client ) => { this . listener . onClientMessage ( client . id , payload . data ) ; } ) . catch ( ( e ) => { console . error ( 'Could not deliver message: ' , e ) ; } ) ; } } , } ; } ; param socket param payload returns _untrustedRequestHandler = ( _socket , payload ) : Partial < Responder < string , any > > => { if ( ! payload . data ) { return { } ; } const clientQuery = JSON . parse ( payload . data ) ; this . listener . onConnectionAttempt ( clientQuery ) ; return { requestResponse : ( payload ) : Single < Payload < string , any > > => { if ( typeof payload . data !== 'string' ) { return new Single ( ( _ ) => { } ) ; } let rawData ; try { rawData = JSON . parse ( payload . data ) ; } catch ( err ) { console . error ( `[conn] Invalid JSON: ${ payload . data }` , 'clientMessage' , 'server' ) ; return new Single ( ( _ ) => { } ) ; } return new Single ( ( subscriber ) => { subscriber . onSubscribe ( undefined ) ; this . _onHandleUntrustedMessage ( clientQuery , rawData ) . then ( ( response ) => { subscriber . onComplete ( { data : response , metadata : '' , } ) ; } ) . catch ( ( err ) => { subscriber . onError ( err ) ; } ) ; } ) ; } , fireAndForget : ( payload ) => { if ( typeof payload . data !== 'string' ) { return ; } let rawData ; try { rawData = JSON . parse ( payload . data ) ; } catch ( err ) { console . error ( `Invalid JSON: ${ payload . data }` , 'server' ) ; return ; } if ( rawData && rawData . method === 'signCertificate' ) { console . debug ( 'CSR received from device' , 'server' ) ; this . _onHandleUntrustedMessage ( clientQuery , rawData ) . then ( ( _ ) => { } ) . catch ( ( err ) => { console . error ( '[conn] Unable to process CSR, failed with error.' , err ) ; } ) ; } } , } ; } ; } export default ServerRSocket ;