Feature/ws api (#72)

initial changes to support websockets as an alternative to webhooks

lib/utils/base-requestor.js

@@ -0,0 +1,75 @@
+const assert = require('assert');
+const Emitter = require('events');
+const crypto = require('crypto');
+const timeSeries = require('@jambonz/time-series');
+let alerter ;
+
+class BaseRequestor extends Emitter {
+  constructor(logger, account_sid, hook, secret) {
+    super();
+    assert(typeof hook === 'object');
+
+    this.logger = logger;
+    this.url = hook.url;
+
+    this.username = hook.username;
+    this.password = hook.password;
+    this.secret = secret;
+    this.account_sid = account_sid;
+
+    const {stats} = require('../../').srf.locals;
+    this.stats = stats;
+
+    if (!alerter) {
+      alerter = timeSeries(logger, {
+        host: process.env.JAMBONES_TIME_SERIES_HOST,
+        commitSize: 50,
+        commitInterval: 'test' === process.env.NODE_ENV ? 7 : 20
+      });
+    }
+  }
+
+  get Alerter() {
+    return alerter;
+  }
+
+  close() {
+    /* subclass responsibility */
+  }
+
+  _computeSignature(payload, timestamp, secret) {
+    assert(secret);
+    const data = `${timestamp}.${JSON.stringify(payload)}`;
+    return crypto
+      .createHmac('sha256', secret)
+      .update(data, 'utf8')
+      .digest('hex');
+  }
+
+  _generateSigHeader(payload, secret) {
+    const timestamp = Math.floor(Date.now() / 1000);
+    const signature = this._computeSignature(payload, timestamp, secret);
+    const scheme = 'v1';
+    return {
+      'Jambonz-Signature': `t=${timestamp},${scheme}=${signature}`
+    };
+  }
+
+  _isAbsoluteUrl(u) {
+    return typeof u === 'string' &&
+      u.startsWith('https://') || u.startsWith('http://') ||
+      u.startsWith('ws://') || u.startsWith('wss://');
+  }
+  _isRelativeUrl(u) {
+    return typeof u === 'string' && u.startsWith('/');
+  }
+  _roundTrip(startAt) {
+    const diff = process.hrtime(startAt);
+    const time = diff[0] * 1e3 + diff[1] * 1e-6;
+    return time.toFixed(0);
+  }
+
+
+}
+
+module.exports = BaseRequestor;

lib/utils/constants.json

@@ -105,6 +105,15 @@
     "Hangup": "hangup",
     "Replaced": "replaced"
   },
+  "HookMsgTypes": [
+    "session:new",
+    "session:reconnect",
+    "session:redirect",
+    "call:status",
+    "queue:status",
+    "verb:hook",
+    "jambonz:error"
+  ],
   "MAX_SIMRINGS": 10,
   "BONG_TONE": "tone_stream://v=-7;%(100,0,941.0,1477.0);v=-7;>=2;+=.1;%(1400,0,350,440)",
   "FS_UUID_SET_NAME": "fsUUIDs"

lib/utils/http-requestor.js

@@ -0,0 +1,105 @@
+const bent = require('bent');
+const parseUrl = require('parse-url');
+const assert = require('assert');
+const BaseRequestor = require('./base-requestor');
+const {HookMsgTypes} = require('./constants.json');
+const snakeCaseKeys = require('./snakecase-keys');
+
+const toBase64 = (str) => Buffer.from(str || '', 'utf8').toString('base64');
+
+function basicAuth(username, password) {
+  if (!username || !password) return {};
+  const creds = `${username}:${password || ''}`;
+  const header = `Basic ${toBase64(creds)}`;
+  return {Authorization: header};
+}
+
+
+class HttpRequestor extends BaseRequestor {
+  constructor(logger, account_sid, hook, secret) {
+    super(logger, account_sid, hook, secret);
+
+    this.method = hook.method || 'POST';
+    this.authHeader = basicAuth(hook.username, hook.password);
+
+    const u = parseUrl(this.url);
+    const myPort = u.port ? `:${u.port}` : '';
+    const baseUrl = this._baseUrl = `${u.protocol}://${u.resource}${myPort}`;
+
+    this.get = bent(baseUrl, 'GET', 'buffer', 200, 201);
+    this.post = bent(baseUrl, 'POST', 'buffer', 200, 201);
+
+
+    assert(this._isAbsoluteUrl(this.url));
+    assert(['GET', 'POST'].includes(this.method));
+  }
+
+  get baseUrl() {
+    return this._baseUrl;
+  }
+
+  /**
+   * Make an HTTP request.
+   * All requests use json bodies.
+   * All requests expect a 200 statusCode on success
+   * @param {object|string} hook - may be a absolute or relative url, or an object
+   * @param {string} [hook.url] - an absolute or relative url
+   * @param {string} [hook.method] - 'GET' or 'POST'
+   * @param {string} [hook.username] - if basic auth is protecting the endpoint
+   * @param {string} [hook.password] - if basic auth is protecting the endpoint
+   * @param {object} [params] - request parameters
+   */
+  async request(type, hook, params) {
+    assert(HookMsgTypes.includes(type));
+    const payload = params ? snakeCaseKeys(params, ['customerData', 'sip']) : null;
+    const url = hook.url || hook;
+    const method = hook.method || 'POST';
+
+    assert.ok(url, 'HttpRequestor:request url was not provided');
+    assert.ok, (['GET', 'POST'].includes(method), `HttpRequestor:request method must be 'GET' or 'POST' not ${method}`);
+    const {url: urlInfo = hook, method: methodInfo = 'POST'} = hook;  // mask user/pass
+    this.logger.debug({url: urlInfo, method: methodInfo, payload}, `HttpRequestor:request ${method} ${url}`);
+    const startAt = process.hrtime();
+
+    let buf;
+    try {
+      const sigHeader = this._generateSigHeader(payload, this.secret);
+      const headers = {...sigHeader, ...this.authHeader};
+      //this.logger.info({url, headers}, 'send webhook');
+      buf = this._isRelativeUrl(url) ?
+        await this.post(url, payload, headers) :
+        await bent(method, 'buffer', 200, 201, 202)(url, payload, headers);
+    } catch (err) {
+      this.logger.error({err, secret: this.secret, baseUrl: this.baseUrl, url, statusCode: err.statusCode},
+        `web callback returned unexpected error code ${err.statusCode}`);
+      let opts = {account_sid: this.account_sid};
+      if (err.code === 'ECONNREFUSED') {
+        opts = {...opts, alert_type: this.Alerter.AlertType.WEBHOOK_CONNECTION_FAILURE, url};
+      }
+      else if (err.name === 'StatusError') {
+        opts = {...opts, alert_type: this.Alerter.AlertType.WEBHOOK_STATUS_FAILURE, url, status: err.statusCode};
+      }
+      else {
+        opts = {...opts, alert_type: this.Alerter.AlertType.WEBHOOK_CONNECTION_FAILURE, url, detail: err.message};
+      }
+      this.Alerter.writeAlerts(opts).catch((err) => this.logger.info({err, opts}, 'Error writing alert'));
+
+      throw err;
+    }
+    const rtt = this._roundTrip(startAt);
+    if (buf) this.stats.histogram('app.hook.response_time', rtt, ['hook_type:app']);
+
+    if (buf && buf.toString().length > 0) {
+      try {
+        const json = JSON.parse(buf.toString());
+        this.logger.info({response: json}, `HttpRequestor:request ${method} ${url} succeeded in ${rtt}ms`);
+        return json;
+      }
+      catch (err) {
+        //this.logger.debug({err, url, method}, `HttpRequestor:request returned non-JSON content: '${buf.toString()}'`);
+      }
+    }
+  }
+}
+
+module.exports = HttpRequestor;

lib/utils/place-outdial.js

@@ -357,7 +357,7 @@ class SingleDialer extends Emitter {
       this.callInfo.updateCallStatus(callStatus, sipStatus);
       if (typeof duration === 'number') this.callInfo.duration = duration;
       try {
-        this.requestor.request(this.application.call_status_hook, this.callInfo.toJSON());
+        this.requestor.request('call:status', this.application.call_status_hook, this.callInfo.toJSON());
       } catch (err) {
         this.logger.info(err, `SingleDialer:_notifyCallStatusChange error sending ${callStatus} ${sipStatus}`);
       }

lib/utils/ws-requestor.js

@@ -0,0 +1,249 @@
+const assert = require('assert');
+const BaseRequestor = require('./base-requestor');
+const short = require('short-uuid');
+const {HookMsgTypes} = require('./constants.json');
+const Websocket = require('ws');
+const snakeCaseKeys = require('./snakecase-keys');
+const HttpRequestor = require('./http-requestor');
+const MAX_RECONNECTS = 5;
+const RESPONSE_TIMEOUT_MS = process.env.JAMBONES_WS_API_MSG_RESPONSE_TIMEOUT || 5000;
+
+class WsRequestor extends BaseRequestor {
+  constructor(logger, account_sid, hook, secret) {
+    super(logger, account_sid, hook, secret);
+    this.connections = 0;
+    this.messagesInFlight = new Map();
+    this.maliciousClient = false;
+
+    assert(this._isAbsoluteUrl(this.url));
+
+    this.on('socket-closed', this._onSocketClosed.bind(this));
+  }
+
+  /**
+   * Send a JSON payload over the websocket.  If this is the first request,
+   * open the websocket.
+   * All requests expect an ack message in response
+   * @param {object|string} hook - may be a absolute or relative url, or an object
+   * @param {string} [hook.url] - an absolute or relative url
+   * @param {string} [hook.method] - 'GET' or 'POST'
+   * @param {string} [hook.username] - if basic auth is protecting the endpoint
+   * @param {string} [hook.password] - if basic auth is protecting the endpoint
+   * @param {object} [params] - request parameters
+   */
+  async request(type, hook, params) {
+    assert(HookMsgTypes.includes(type));
+    const url = hook.url || hook;
+
+    if (this.maliciousClient) {
+      this.logger.info({url: this.url}, 'WsRequestor:request - discarding msg to malicious client');
+      return;
+    }
+
+    /* if we have an absolute url, and it is http then do a standard webhook */
+    if (this._isAbsoluteUrl(url) && url.startsWith('http')) {
+      this.logger.debug({hook}, 'WsRequestor: sending a webhook');
+      const requestor = new HttpRequestor(this.logger, this.account_sid, hook, this.secret);
+      return requestor.request(type, hook, params);
+    }
+
+    /* connect if necessary */
+    if (!this.ws) {
+      if (this.connections >= MAX_RECONNECTS) {
+        throw new Error(`max attempts connecting to ${this.url}`);
+      }
+      try {
+        const startAt = process.hrtime();
+        await this._connect();
+        const rtt = this._roundTrip(startAt);
+        this.stats.histogram('app.hook.connect_time', rtt, ['hook_type:app']);
+      } catch (err) {
+        this.logger.error({err}, 'WsRequestor:request - failed connecting');
+        throw err;
+      }
+    }
+    assert(this.ws);
+
+    /* prepare and send message */
+    const payload = params ? snakeCaseKeys(params, ['customerData', 'sip']) : null;
+    assert.ok(url, 'WsRequestor:request url was not provided');
+
+    const msgid = short.generate();
+    const obj = {
+      type,
+      msgid,
+      hook: type === 'verb:hook' ? url : undefined,
+      data: {...payload}
+    };
+
+    this.logger.debug({obj}, `WsRequestor:request ${url}`);
+
+    /* simple notifications */
+    if (['call:status', 'jambonz:error'].includes(type)) {
+      this.ws.send(JSON.stringify(obj));
+      return;
+    }
+
+    /* messages that require an ack */
+    return new Promise((resolve, reject) => {
+      /* give the far end a reasonable amount of time to ack our message */
+      const timer = setTimeout(() => {
+        const {failure} = this.messagesInFlight.get(msgid);
+        failure && failure(`timeout from far end for msgid ${msgid}`);
+        this.messagesInFlight.delete(msgid);
+      }, RESPONSE_TIMEOUT_MS);
+
+      /* save the message info for reply */
+      const startAt = process.hrtime();
+      this.messagesInFlight.set(msgid, {
+        success: (response) => {
+          clearTimeout(timer);
+          const rtt = this._roundTrip(startAt);
+          this.logger.info({response}, `WsRequestor:request ${url} succeeded in ${rtt}ms`);
+          this.stats.histogram('app.hook.ws_response_time', rtt, ['hook_type:app']);
+          resolve(response);
+        },
+        failure: (err) => {
+          clearTimeout(timer);
+          reject(err);
+        }
+      });
+
+      /* send the message */
+      this.ws.send(JSON.stringify(obj));
+    });
+  }
+
+  close() {
+    this.logger.info('WsRequestor: closing socket');
+    if (this.ws) {
+      this.ws.close();
+      this.ws.removeAllListeners();
+    }
+  }
+
+  _connect() {
+    assert(!this.ws);
+    return new Promise((resolve, reject) => {
+      let opts = {
+        followRedirects: true,
+        maxRedirects: 2,
+        handshakeTimeout: 1000,
+        maxPayload: 8096,
+      };
+      if (this.username && this.password) opts = {...opts, auth: `${this.username}:${this.password}`};
+
+      this
+        .once('ready', (ws) => {
+          this.ws = ws;
+          this.removeAllListeners('not-ready');
+          resolve();
+        })
+        .once('not-ready', () => {
+          this.removeAllListeners('ready');
+          reject();
+        });
+      const ws = new Websocket(this.url, ['ws.jambonz.org'], opts);
+      this._setHandlers(ws);
+    });
+  }
+
+  _setHandlers(ws) {
+    ws
+      .on('open', this._onOpen.bind(this, ws))
+      .on('close', this._onClose.bind(this))
+      .on('message', this._onMessage.bind(this))
+      .on('unexpected-response', this._onUnexpectedResponse.bind(this, ws))
+      .on('error', this._onError.bind(this));
+  }
+
+  _onError(err) {
+    this.logger.info({url: this.url, err}, 'WsRequestor:_onError');
+    if (this.connections > 0) this.emit('socket-closed');
+    this.emit('not-ready');
+  }
+
+  _onOpen(ws) {
+    assert(!this.ws);
+    this.emit('ready', ws);
+    this.logger.info({url: this.url}, 'WsRequestor - successfully connected');
+  }
+
+  _onClose() {
+    this.logger.info({url: this.url}, 'WsRequestor - socket closed unexpectedly from remote side');
+    this.emit('socket-closed');
+  }
+
+  _onUnexpectedResponse(ws, req, res) {
+    assert(!this.ws);
+    this.logger.info({
+      headers: res.headers,
+      statusCode: res.statusCode,
+      statusMessage: res.statusMessage
+    }, 'WsRequestor - unexpected response');
+    this.emit('connection-failure');
+    this.emit('not-ready');
+  }
+
+  _onSocketClosed() {
+    this.ws = null;
+    if (this.connections > 0) {
+      if (++this.connections < MAX_RECONNECTS) {
+        setImmediate(this.connect.bind(this));
+      }
+    }
+  }
+
+  _onMessage(content, isBinary) {
+    if (this.isBinary) {
+      this.logger.info({url: this.url}, 'WsRequestor:_onMessage - discarding binary message');
+      this.maliciousClient = true;
+      this.ws.close();
+      return;
+    }
+
+    /* messages must be JSON format */
+    try {
+      const {type, msgid, command, queueCommand = false, data} = JSON.parse(content);
+      assert.ok(type, 'type property not supplied');
+
+      switch (type) {
+        case 'ack':
+          assert.ok(msgid, 'msgid not supplied');
+          this._recvAck(msgid, data);
+          break;
+
+        case 'command':
+          assert.ok(command, 'command property not supplied');
+          assert.ok(data, 'data property not supplied');
+          this._recvCommand(msgid, command, queueCommand, data);
+          break;
+
+        default:
+          assert.ok(false, `invalid type property: ${type}`);
+      }
+    } catch (err) {
+      this.logger.info({err}, 'WsRequestor:_onMessage - invalid incoming message');
+    }
+  }
+
+  _recvAck(msgid, data) {
+    const obj = this.messagesInFlight.get(msgid);
+    if (!obj) {
+      this.logger.info({url: this.url}, `WsRequestor:_recvAck - ack to unknown msgid ${msgid}, discarding`);
+      return;
+    }
+    this.logger.debug({url: this.url}, `WsRequestor:_recvAck - received response to ${msgid}`);
+    this.messagesInFlight.delete(msgid);
+    const {success} = obj;
+    success && success(data);
+  }
+
+  _recvCommand(msgid, command, queueCommand, data) {
+    // TODO: validate command
+    this.logger.info({msgid, command, queueCommand, data}, 'received command');
+    this.emit('command', {msgid, command, queueCommand, data});
+  }
+}
+
+module.exports = WsRequestor;