host comms with validation

Author: sevenbitbyte

src/comms/host-op.js src/comms/host/host-op.jssrc/comms/host-op.js renamed to src/comms/host/host-op.js

@@ -1,4 +1,4 @@
-const reach = require('../utils/reach')
+const reach = require('../../utils/reach')
 const debug = require('debug')('dataparty.op.host-op')
 const moment = require('moment')
 const EventEmitter = require('eventemitter3')
@@ -34,6 +34,7 @@ class Op extends EventEmitter {
     this.end = undefined
     this.state = Op.STATES.Pending
     this.level = Op.STATUS_LEVELS.Status_Info
+    this.result = null
   }
 
   get op(){

src/comms/protocol-scheme.js src/comms/host/host-protocol-scheme.jssrc/comms/protocol-scheme.js renamed to src/comms/host/host-protocol-scheme.js

@@ -5,28 +5,29 @@ const ID_SCHEME = Joi.alternatives().try(Joi.string(), Joi.number())
 
 const OP_HEADER = Joi.object().keys({
   id: ID_SCHEME.required(),
-  op: Joi.string().valid([
+  op: Joi.string().valid(
     'auth',
-    'call',
+    'peer-call',
     'advertise',
     'subscribe',
     'unsubscribe',
     'publish'
-  ]).required()
+  ).required()
 })
 
 
 const AUTH_OP = Joi.object().keys({
   id: ID_SCHEME.required(),
   op: Joi.string().valid('auth').required(),
-  session: Joi.objectId().required()
+  session: Joi.string().required(), //Joi.objectId().required(),
 })
 
 const CALL_OP = Joi.object().keys({
   id: ID_SCHEME.required(),
-  op: Joi.string().valid('call').required(),
+  op: Joi.string().valid('peer-call').required(),
+  session: Joi.string(), //Joi.objectId(),
   endpoint: Joi.string(),
-  body: Joi.string()
+  data: Joi.object()
 })
 
 const ADVERTISE_OP = Joi.object().keys({
@@ -75,7 +76,8 @@ const ANY_OP = Joi.alternatives().try(
   UNADVERTISE_OP,
   SUBSCRIBE_OP,
   UNSUBSCRIBE_OP,
-  PUBLISH_OP
+  PUBLISH_OP,
+  CALL_OP
 )
 
 module.exports = {
@@ -86,5 +88,6 @@ module.exports = {
   'SUBSCRIBE_OP': SUBSCRIBE_OP,
   'UNSUBSCRIBE_OP': UNSUBSCRIBE_OP,
   'PUBLISH_OP': PUBLISH_OP,
+  'PEER_CALL_OP': CALL_OP,
   'ANY_OP': ANY_OP
 }

src/comms/peer-comms.js

@@ -4,29 +4,127 @@ const debug = require('debug')('dataparty.comms.peercomms')
 const SocketOp = require('./op/socket-op')
 const SocketComms = require('./socket-comms')
 
+const Joi = require('@hapi/joi')
+const HostOp = require('./host/host-op')
+const HostProtocolScheme = require('./host/host-protocol-scheme')
+
 const AUTH_TIMEOUT_MS = 3000
 
+const HOST_SESSION_STATES = {
+  AUTH_REQUIRED: 'AUTH_REQUIRED',
+  AUTHED: 'AUTHED',
+  SERVER_CLOSED: 'SERVER_CLOSED',
+  CLIENT_CLOSED: 'CLIENT_CLOSED'
+}
+
 class PeerComms extends SocketComms {
   constructor({remoteIdentity, discoverRemoteIdentity, host, party, socket, ...options}){
     super({remoteIdentity, discoverRemoteIdentity, party, ...options})
 
     this.socket = socket || null
 
-    this.host = host   //! Is comms host
-    this.party = party
+    this.host = host   //! Is comms host\
     this.oncall = null
 
     this._host_auth_timeout = null
 
+    if(this.host){
+      this.state = PeerComms.STATES.AUTH_REQUIRED
+      this.session = undefined
+      this.identity = undefined
+      this.actor = undefined
+    }
+
     this.pending_calls = 0
   }
 
+  setState(state) {
+    this.state = state
+    this.emit('state', this.state)
+  }
+
+  static get STATES() {
+    return HOST_SESSION_STATES
+  }
+
   async handleClientCall(message){
+
+    debug('handleClientCall - pending calls - ', this.pending_calls)
+
+    this.pending_calls++
+    try{
+
+      let response = null
+      const request = await this.decrypt( {data: message}, this.remoteIdentity )
+      debug('handleHostCall', request)
+
+      let inputValidated
+
+      if (this.state === PeerComms.STATES.AUTHED) {
+        debug('handling authed call')
+        inputValidated = HostProtocolScheme.ANY_OP.validate(request)
+      } else if (this.state === PeerComms.STATES.AUTH_REQUIRED) {
+        debug('handling non-authed call')
+        inputValidated = HostProtocolScheme.AUTH_OP.validate(request)
+      } else {
+        throw new Error(
+          'Recieved input in unexpected session state [',
+          this.state,
+          ']'
+        )
+      }
+
+      if(inputValidated.error !== undefined){
+        throw inputValidated.error
+      }
+
+      debug('original input ->', typeof request, request)
+      debug('validated input ->', inputValidated)
+      const op = new HostOp({ msg: message, input: inputValidated.value })
+
+      /*debug('session id : ', op.input.session, this.session)
+
+      if (this.session && op.input.session === this.session.id) {
+        debug('session id MATCH')
+      }*/
+
+      op.once('finished', state => {
+        const response = {
+          op: 'status',
+          id: op.id,
+          level: op.level,
+          state: op.state,
+          stats: {
+            start: op.start,
+            end: op.end,
+            duration_ms: op.end - op.start
+          },
+          ...op.result 
+        }
+
+        this.send(response)
+      })
+
+
+      
+      await this.authorizeOperation(op)
+
+    } catch (err) {
+      debug('EXCEPTION ->', err)
+    }
+    this.pending_calls--
+
+
+    return
+
     let response = null
     const request = await this.decrypt( {data: message}, this.remoteIdentity )
     debug('handleHostCall', request)
 
     if(!this.authed){
+
+
+
       if(request.op == 'auth'){
         debug('allowing client')
         response = {
@@ -66,7 +164,7 @@ class PeerComms extends SocketComms {
 
 
   async handleClientConnection(){
-    debug('handleHostConnection')
+    debug('handleClientConnection')
 
     this._host_auth_timeout = setTimeout(
       this.handleAuthTimeout.bind(this),
@@ -90,8 +188,8 @@ class PeerComms extends SocketComms {
     this.onmessage({data: message})
   }
 
-  async call(path, data){
-    if(this.host){ throw new Error('host-not-allowed-call') }
+  async call(path, data, force=false){
+    if(this.host && !this.force){ throw new Error('host-not-allowed-call') }
     if(!this.authed){ throw new Error('not authed') }
 
     if (!this.party.hasIdentity()) {
@@ -115,6 +213,7 @@ class PeerComms extends SocketComms {
 
     if(this.host){
       debug('host mode comms')
+
       this.socket.on('connect', this.handleClientConnection.bind(this))
       this.socket.on('data', this.handleClientCall.bind(this))
     }
@@ -136,10 +235,57 @@ class PeerComms extends SocketComms {
   close(){
     debug('closing connection')
     this.socket.destroy()
-}
+  }
+
+
+  async authorizeOperation(op) {
+
+    debug('Here\'s op', op)
+    
+    debug('Here state : ', this.state)
+
+    if (op.op === 'auth' && this.state === PeerComms.STATES.AUTH_REQUIRED) {
+    
+      debug('handling auth op')
+      return this.handleAuthOp(op)
+    
+    } else if (op.op === 'peer-call' && this.state === PeerComms.STATES.AUTHED) {
+
+      return this.handleCallOp(op)
 
+    } else {
+      op.result='not implemented'
+      op.setState(HostOp.STATES.Finished_Fail)
+    }
+  }
+
+  async handleAuthOp(op){
+    
+    debug('allowing client - ', this.remoteIdentity)
+
+    clearTimeout(this._host_auth_timeout)
+    this._host_auth_timeout = null
 
+    this.authed = true
+    this.setState(PeerComms.STATES.AUTHED)
+    op.setState(HostOp.STATES.Finished_Success)
 
+    this.emit('open')
+    return 
+  }
+
+  async handleCallOp(op){
+    debug('peer-call')
+    if(op.input.endpoint == 'api-v2-peer-bouncer'){
+      
+      debug('ask->',op.input.data)
+      op.result = await this.party.handleCall(op.input.data)
+
+      op.setState(HostOp.STATES.Finished_Success)
+
+      return
+    }
+  }
 }